标签: sqlalchemy

我在一个现有的mysql数据库中反映了一大堆表.我想表示任何表中某个名称的任何列都默认为datetime.now().然而,天真地循环遍历表和列,只是在我发现具有某个名称的那些上设置默认值不起作用,在执行session.add()时; session.flush()我收到以下错误:

AttributeError: 'builtin_function_or_method' object has no attribute '__visit_name__'

这似乎与调用_set_parent(以及sqlalchemy.schema self._init_items(*toinit)中第721行的调用)有关.

有没有人知道是否有办法做到这一点,没有经过我所有的反射表,并添加列(..)行所有做同样的事情或诉诸真正丑陋的黑客？

我在数据库中的表如下

Username    city        Type 
 Anna       Paris        abc 
 Marc       london       abc
 erica      rome         AF 
 Sara       Newyork      cbd 
 silvia      paris        AD

我有一个包含字符串值的列表

typelist = {'abc', 'cbd'}

我想使用 sqlalchemy 查询数据库，从列类型等于列表中的值的表中获取数据：

Username    city        Type 
 Anna       Paris        abc 
 Marc       london       abc
 Sara       Newyork      cbd 

我正在尝试这段代码

sql = "SELECT * FROM table WHERE data IN :values" 
query = sqlalchemy.text(sql).bindparams(values=tuple(typelist))
conn.engine.execute(query)

但它只返回类型列表中的一个值，而不是所有列表值。

Username    city        Type 
 Sara       Newyork      cbd 

我试图让 Marshmallow-SQLAlchemy 反序列化具有嵌套对象的对象，而不指定嵌套对象的外键（应该是父对象的主键）。这是一个独立的示例：

# Python version == 3.8.2
from datetime import datetime
import re

# SQLAlchemy == 1.3.23
from sqlalchemy import func, create_engine, Column, ForeignKey, Text, DateTime
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy.orm import relationship, sessionmaker

# marshmallow==3.10.0
# marshmallow-sqlalchemy==0.24.2
from marshmallow import fields
from marshmallow.fields import Nested
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema

################################################################################
# Set up
################################################################################

engine = create_engine("sqlite:///test.db")

Session = sessionmaker()
Session.configure(bind=engine)
session = Session()


################################################################################
# Models
################################################################################

@as_declarative()
class Base(object):

    @declared_attr
    def __tablename__(cls):
        # From …

如果我的问题很平庸，我提前道歉：我是 SQL 的初学者。

我想创建一个简单的数据库，有两个表：Students和Answers。基本上，每个学生将回答三个问题（每个问题可能的答案是True或False），他的答案将存储在Answers表中。 Students可以有两个“经验”级别：“本科生”和“研究生”。获得“研究生”经验级别Answers所提供的所有内容的最佳方式是什么？Students

这就是我为Students和Answers表中的条目定义 SQLAlchemy 类的方式：

import random

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, Date, Boolean, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship


db_uri = "sqlite:///simple_answers.db"
db_engine = create_engine(db_uri)
db_connect = db_engine.connect()

Session = sessionmaker()
Session.configure(bind=db_engine)
db_session = Session()


Base = declarative_base()    


class Student(Base):
    __tablename__ = "Students"
    
    id = Column(Integer, primary_key=True)
    experience = …

我有这样一个案例：

|               Note table               |
|---------------------|------------------|
|          id         |     parent_id    |
|---------------------|------------------|
|          1          |     Null         |
|---------------------|------------------|
|          2          |      1
|---------------------|------------------|
|          3          |      2
|---------------------|------------------|
|          4          |      3
|---------------------|------------------|

我想要实现的是获取顶级父 ID。在这种情况下，如果我传递 ID 号 4，我将得到 Id 1，因为 Id 1 是顶级父级。当parent_id 达到 null 时，意味着该 id 是顶级父级。

我已经尝试过了，但返回的是我传递给函数的 Id。

  def get_top_level_Note(self, id: int):


        hierarchy = self.db.session.query(Note).filter(Note.id == id).cte(name="hierarchy", recursive=True)

        parent = aliased(hierarchy, name="p")
        children = aliased(Note, name="c")

        hierarchy = hierarchy.union_all(self.db.session.query(children).filter(children.parent_id == parent.c.id))

        result = self.db.session.query(Note).select_entity_from(hierarchy).all()

因此，我立即将一个数据列表插入到 postgres 中：

session.bulk_insert_mappings(model_name, data)
session.commit()

但是，执行此操作时，不会触发事件“ after_insert ”。Sqlalchemy 中的事件是否不会触发批量操作，还是我遗漏了某些内容？我是 Sqlalchemy 中事件处理的新手

这里我再次尝试学习SqlAlchemy。

我正在执行插入过程，但最后，SqlAlchemy 正在回滚。我不明白为什么会发生这种情况。

这是 I\xc2\xb4ve 创建的表：

NiceSchools = Table(\n   'NiceSchools', metadata, \n   Column('id', Integer, primary_key = True, autoincrement = True), \n   Column('name', String (50), index = True, nullable = False), \n   Column('Safety_Score', Float, nullable = False),\n   Column('Location', String(50), nullable = False), \n   Column('Start_date', DateTime, default = datetime.today())\n)\n

这是插入语句：

stmt = NiceSchools.insert().values(id = 1, name = 'F\xc3\xa1bio', Safety_Score = 80, Location = 'Sa\xc3\xbade')\nwith engine.connect() as conn:\n    conn.execute(stmt)\n

这是 echo=true 日志：

2022-06-21 16:59:23,798 INFO sqlalchemy.engine.Engine …

我使用 sqlalchemy 库将 JSON 字符串添加到 python 中的 Postgres jsonb 列。

创建 json 字符串的代码是

survey_data_series = scto_df.loc[int(i)]

survey_data_json_string = json.dumps(survey_data_series.to_dict())

添加到 postgresql 数据库的代码是：

def add_record(self, table_name, record_data):
    record = getattr(self, table_name)(**record_data)
    self.s.add(record)
    self.s.flush()
    return(record.id) 

我的 JSON 在日志中正确打印

{"completiondate": "Feb 8, 2022 10:49:29 AM", "submissiondate": "Feb 8, 2022 10:49:29 AM", "starttime": "Feb 8, 2022 10:37:17 AM", "endtime": "Feb 8, 2022 10:49:21 AM", "deviceid": "ac29d854b49a49be", "subscriberid": NaN}

但是当它进入数据库时​​，json 被包含在一个字符串中，并且在每个双引号前面添加一个转义字符。

"{\"completiondate\": \"Feb 8, 2022 10:49:29 AM\", \"submissiondate\": \"Feb 8, 2022 10:49:29 AM\", \"starttime\": \"Feb …

当我尝试执行下面的代码时，mariadb.ProgrammingError: Cursor is closed即使数据已成功写入数据库，python 也会给我一个 。我猜想返回行计数有问题（请参阅下面的完整回溯）。这是一个可能的错误还是我错过了什么？

代码示例

import pandas as pd
from sqlalchemy import create_engine
conn_str = 'mariadb+mariadbconnector://user:password@localhost:3306/database_name'
engine = create_engine(conn_str)
df = pd.DataFrame(data={'col1': [1, 2, 3], 'col2': ['A', 'B', 'C']})
df.to_sql('pandas_test_table', index=False, con=engine, if_exists='replace')

追溯

Traceback (most recent call last):
  File "C:\Program Files\Python310\lib\code.py", line 90, in runcode
    exec(code, self.locals)
  File "<input>", line 1, in <module>
  File "C:\Users\user\AppData\Roaming\Python\Python310\site-packages\pandas\core\generic.py", line 2951, in to_sql
    return sql.to_sql(
  File "C:\Users\user\AppData\Roaming\Python\Python310\site-packages\pandas\io\sql.py", line 697, in to_sql
    return pandas_sql.to_sql(
  File "C:\Users\user\AppData\Roaming\Python\Python310\site-packages\pandas\io\sql.py", line 1739, in to_sql …

我正在尝试使用 python-oracledb 库而不是 cx_Oracle 创建引擎，但它显示未找到 cx_Oracle 模块。有什么方法可以让 sqlalchemy 使用 python-oracledb 创建引擎吗？

如果我使用 cx_Oracle，我曾经使用以下内容创建一个引擎：

conn_str = "oracle+cx_oracle://{user}:{password}@{host}:{port}"

我正在使用下面的代码来使其使用 python-oracledb。

import oracledb
from sqlalchemy import create_engine, text

oracledb.init_oracle_client()

engine = create_engine(f'oracle://{p_username}:{p_password}@{p_dns}:{p_port}', max_identifier_length=128)

print(engine)