标签: sqlalchemy

我正在开发一个使用 Flask、SQLAlchemy 和 PostgreSQL 的应用程序。我必须编写一个在数据库上执行多个查询的事务。

def exec_query_1():
    with db.engine.connect() as connection:
        connection.execute(#some-query)

def exec_query_2():
    with db.engine.connect() as connection:
        connection.execute(#some-query)

def exec_query_3():
    with db.engine.connect() as connection:
        connection.execute(#some-query)

def execute_transaction():
    with db.engine.connect() as connection:
        with connection.begin() as transaction:
            exec_query_1()
            exec_query_2()
            exec_query_3()

鉴于应用程序是多线程的，这段代码会按预期工作吗？如果是，怎么办？如果不是，那么使其发挥作用的正确方法是什么？

backref有没有办法从 SqlAlchemy 中的关系中检索元素子集，但保持创建类时我在（关系）中定义的顺序？

假设我有两个表/模型： AUser和 a UserLog，其中包含用户执行的操作（只是Basedeclarative_base ()）

每个日志UserLog都有一个执行操作的 ID 的外键User，并且每个日志的操作类型可以从Enum. 每个用户日志还有一个created我订购的时间戳：在常规情况下，我希望从最旧到最新（最旧的在前）获取日志。

但是，我还希望能够让特定用户仅加载日志的子集（特定操作的日志）。我知道这违背了连接负载的“要点”，因为我会得到不切实际的数据库状态（有更多日志链接到我获取的对象User将反映的用户），但在某些情况下它会很有帮助。

class UserLog(Base):
    __tablename__ = 'user_logs'
    id = Column(UUID(as_uuid=True), primary_key=True)
    timestamp_created = Column(
        DateTime, server_default=sqlalchemy.text("now()")
    )
    user_id = Column(
        UUID(as_uuid=True),
        ForeignKey("users.id", ondelete="CASCADE")
    )
    action = Column(SaEnum(UserLogsConstants), nullable=False)

    user = relationship(
        "User", 
        backref=backref(
          'logs', order_by="asc(UserLog.timestamp_created)"
        ) # The order_by is important here
    )

感谢这个 SO 答案，我已经能够使用 加载日志子集contains_eager，但随后我丢失了timestamp_created …

我正在尝试启动与 Azure SQL 数据库的连接。出于安全原因，我无法硬编码用户名和密码，我需要从配置文件中的字典中获取它，并以这种方式将其放入连接 URL 中。

当我尝试这样做并在 Jupyter 笔记本中运行它时， 并{key[value]}没有被配置文件中引用的内容替换。

如果我使用格式化字符串，我会得到一个KeyError. 如果我不使用格式化字符串，它只会打印{variablename}字面意思。

我对我的 SQLAlchemy 模型做了一些更改，我想将它们迁移到我的数据库。当我尝试迁移时，出现以下错误：

$ flask db migrate\nUsage: flask db migrate [OPTIONS]\n\nError: While importing "minerva.dashboard.app", an ImportError was raised:\n\nTraceback (most recent call last):\n  File "/home/mattfred/Minerva/envs/dashboard/lib/python3.7/site-packages/flask/cli.py", line 235, in locate_app\n    __import__(module_name)\n  File "/home/mattfred/Minerva/minerva/dashboard/app.py", line 13, in <module>\n    from templates.nav_layout import *\nModuleNotFoundError: No module named \'templates\'\n

我不知道自从我第一次初始化数据库以来发生了什么变化，除了将视图移动到单独的文件而不是 1 中。

我实际上可以通过将导入语句更改为来解决此问题

from .templates.nav_layout import *\n

但当我尝试运行该应用程序时出现此错误：

Traceback (most recent call last):\n  File "app.py", line 13, in <module>\n    from .templates.nav_layout import *\nModuleNotFoundError: No module named \'__main__.templates\'; \'__main__\' is not …

我想使用 SQLAlchemy 提供的 ON DUPLICATE KEY UPDATE 选项来更新插入一堆记录。

这些记录已使用以下命令成功插入Python（其中connection是engine.connect()对象，table是Table对象）

record_list = [{'col1': 'name1', 'col2': '2015-01-31', 'col3': 27.2},
               {'col1': 'name1', 'col2': '2016-01-31', 'col3': 25.2}]
query = insert(table)
results = connection.execute(query, record_list)

查看https://docs.sqlalchemy.org/en/13/dialects/mysql.html#insert-on-duplicate-key-update-upsert上的文档以及许多 SO 问题（包括可能的建议在SQLAlchemy ON DUPLICATE KEY UPDATE的评论下）我尝试了许多不同的示例，但没有一个我可以看到使用此方法通过 upsert 语句处理多个记录。

我正在尝试沿着

query = insert(table).values(record_list)
upsert_query = query.on_duplicate_key_update()
results = connection.execute(upsert_query)

但要么遇到 .on_duplicate_key_update() 要求不能为空的问题，要么 SQL 语法错误。

如果有人成功管理并可以帮助我处理这里的代码结构，我将非常感激。

我正在尝试在 SQL DB 中更新并添加新的数据框架 pandas 数据框架。

我有两个查询：一个是导入 DF 中的所有数据（超过 100.000），并使用以下代码将其与 sql 表进行比较：

df.to_sql(table_name, con=engine, if_exists='replace', index=False)

第二个是相同的导入和查询，但只是将特定时间段内的数据导入到数据框中，并将其导入到同一个sql表中。使用的代码是相同的：

 df.to_sql(table_name, con=engine, if_exists='replace', index=False)

我的问题是：当我使用第二个代码时，它删除了 sql 表中不存在于我的第二个代码（部分导入）中的所有现有数据。

有人可以给我建议吗？

有关信息，ma 数据库位于 Azure 上

谢谢，新年快乐

我对 python 很陌生。我正在使用 Flask 使用 Python 开发微服务。为了连接到数据库，我使用 Sqlalchemy。是否有任何标准方法，例如创建模型并将它们连接到数据库中的现有表？与 Spring boot 服务一样，我们是否有用于查询的存储库类或者标准过程是什么。任何代码片段表示赞赏。

我一直试图找出导致此输入类型错误的原因，但似乎无法弄清楚。

模型/pin.py

from sqlalchemy import Column, String


class Pin(db.Model):
    __tablename__ = 'pin'

    pin = Column(String, primary_key=True)

模式/pin.py

from ..models.pin import Pin
from marshmallow import fields, validate


class PinSchema(ma.Schema):
    pin = fields.Str(required=True, validate=[validate.Length(5)])

    class Meta:
        model = Pin
        fields = ('pin')

pin_schema = PinSchema()

资源/pin.py

from flask_restful import Resource
from flask import request
from ..schemas.pin import pin_schema


class PinResource(Resource):

    """
        Update current pin
        Expecting:
        {
            'pin': 'current_pin'
            'new_pin': 'new_pin'
        }
    """
    def patch(self):
        
        data = request.get_json()
        
        errors = pin_schema.validate(data['new_pin'])

        if errors: …

下面的代码是如何工作的？

result = session.query(Customers).filter(Customers.id == 2)

根据我不扎实的 Python 知识，Customers.id == 2函数调用中使用的语法不是有效的吗？或者这样的foo == bar语法将首先将 eval 转换为布尔值，然后调用.filter(True/False).

我需要能够在sqlalchemy. 这两列可能来自两个不同的表。该操作的纯 SQL 模拟是：

Select Id,
       Case When Col1 < Col2 Then Col1
            Else Col2
            End As TheMin
From   TableName

我一直在浏览文档，但没有找到直接的方法来做到这一点。这可以如何实施sqlalchemy？