我收到这个警告：
/app/facelo/extensions.py:36：SAWarning：表“试验”上的 DELETE 语句预计会删除 1 行；0 条已匹配。请在映射器配置中设置confirm_deleted_rows=False 以防止出现此警告。
我想忽略它。我尝试使用：
@pytest.mark.filterwarnings("ignore::SAWarning")
它给了我：
INTERNALERROR> warnings._OptionError: unknown warning category: 'SAWarning'
所以它不起作用，因为 pytest 不知道这个警告。如何忽略这个警告？
我根据教程在 python 中创建了一个 sqlite 数据库：
from sqlalchemy import create_engine
engine = create_engine('sqlite:///DB.db')
下次我运行 python 脚本时，它里面有旧数据。
如何删除所有行、重新初始化甚至删除数据库（它位于哪里？）？
我有一个单线程进程，它使用 SQLAlchemy 运行一堆查询。一段时间后，它要么开始进展非常缓慢，要么开始出现以下错误：
(<class 'sqlalchemy.exc.TimeoutError'>): QueuePool limit of size 10 overflow 0 reached, connection timed out, timeout 30 (Background on this error at: http://sqlalche.me/e/3o7r)
我读过那些文档。由于我没有并发连接，似乎只有当我不调用时才会发生这种情况session.close()，我已经验证了这一点。
我添加了一些日志记录来尝试了解发生了什么。我定期记录session.get_bind().pool.status()，这是我所看到的示例：
size: 10 Connections in pool: 0 Current Overflow: -9 Current Checked out connections: 1
size: 10 Connections in pool: 4 Current Overflow: -2 Current Checked out connections: 4
size: 10 Connections in pool: 4 Current Overflow: 0 Current Checked out connections: 6
size: 10 Connections in pool: 3 Current Overflow: …我想导入一个大的 csv 文件，而不需要先将该 csv 文件创建表到 postgresql 中。在 stackoverflow 上搜索时，一位用户建议使用以下代码：
from sqlalchemy import create_engine
engine = create_engine('postgresql://username:password@localhost:5432/mydatabase')
df.to_sql('table_name', engine)
基于此，我根据我的 postgresql 的信息编写了以下代码，其中用户名 =“postgres”，密码 =“Katherine@412”：
from sqlalchemy import create_engine
engine = create_engine('postgresql://[postgres]:Katherine@412@postgres:5432/covid_deaths.csv')
df.to_sql('covid_deaths_owid', engine)
但是它给了我以下错误：
OperationalError: (psycopg2.OperationalError) could not translate host name "412@postgres" to address: Unknown server error
有人可以告诉我我做错了什么吗？
我正在尝试为基于模型的表创建多个（即2列索引）。但我不想给出这个索引的具体名称。我想要这样naming_convention，并且alembic revision --autogenerate会通过命名索引来完成它的工作。到目前为止我有这样的代码：
from sqlalchemy import MetaData
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.schema import Index
metadata = MetaData(
naming_convention={
'pk': '%(table_name)s_pk',
'ix': '%(table_name)s_%(column_0_N_name)s_ix',
},
)
@as_declarative(metadata=metadata)
class Base:
pass
class Foo(Base):
id = Column(Integer, primary_key=True)
col1 = Column('Col1', Integer)
col2 = Column('Col2', DateTime)
Index(
metadata.naming_convention['ix'] % {
'table_name': Foo.__tablename__,
'column_0_N_name': Foo.col1.expression.name + "_" + Foo.col2.expression.name
},
Foo.col1,
Foo.col2,
)
所以我想避免代码的“创建名称”部分：
metadata.naming_convention['ix'] % {
'table_name': Foo.__tablename__,
'column_0_N_name': Foo.col1.expression.name + "_" + Foo.col2.expression.name
}
如果您在 SQL Alchmey 中创建了一个实体，您可以在稍后获取您指定的列名称以在自定义语句中使用。
例如，这是数据实体的第一部分：
class ReportMetadataEntity(Base, ReportMetadata):
__tablename__ = "reports_metadata"
__mapper_args__ = {"eager_defaults": True}
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, name="id")
application_customer_id = Column(
String,
nullable=False,
name="application_customer_id",
)
有没有办法调用 ReportMetadataEntity.id.name 或其他一些属性来获取字符串“id”？
谢谢
根据文档，我们应该对对象使用以下模式sessionmaker：
Session = sessionmaker(engine)
with Session.begin() as session:
session.add(some_object)
在多线程环境中，我们也应该使用单线程scoped_session并共享它。因此，在我的程序中__init__.py，我创建了一个并将其导入到程序中的其他位置：
engine = create_engine(config.SQLALCHEMY_DATABASE_URI)
Session = scoped_session(sessionmaker(bind=engine))
问题是，我应该如何结合这两种方法？这似乎是建议的方式，但它出错了：
from myapp import Session
with Session.begin() as session:
query_result = session.query(MyModel).all()
----
Exception has occurred: AttributeError
'SessionTransaction' object has no attribute 'query'
我尝试了以下方法并且它有效，但它似乎不遵循文档，而且我担心它会破坏一些不明显的东西。谁能确认这是否正确？
from myapp import Session
with Session() as session, session.begin():
query_result = session.query(MyModel).all()
我一直在查看其他回复，但几乎没有看到解决具体问题的答案。
我之前使用 SQLAlchemy 1.4.44 和 pandas 1.5.1，并且执行 SQL 存储过程的以下代码有效：
sql_connection: str = "driver={ODBC Driver 17 for SQL Server};server=localhost\SQLEXPRESS;database=Finance;trusted_connection=yes"
sql_engine: sqlalchemy.engine = sqlalchemy.create_engine(
url=sqlalchemy.engine.URL.create(drivername="mssql+pyodbc", query={"odbc_connect": sql_connection})
)
with sql_engine.connect() as connection:
query: str = "EXEC dbo.get_latest_tickers @etf=?"
return pandas.read_sql_query(sql=query, con=connection, params=[etf])
我切换到 SQLAlchemy 2.0.3 和 pandas 1.5.3，并通过使用 sqlalchemy.text 包装对存储过程的调用来修改代码，因为此版本的 SQLAlchemy 需要它。修改后的代码如下：
sql_connection: str = "driver={ODBC Driver 17 for SQL Server};server=localhost\SQLEXPRESS;database=Finance;trusted_connection=yes"
sql_engine: sqlalchemy.engine = sqlalchemy.create_engine(
url=sqlalchemy.engine.URL.create(drivername="mssql+pyodbc", query={"odbc_connect": sql_connection})
)
with sql_engine.connect() as connection:
query: str = sqlalchemy.text("EXEC dbo.get_latest_tickers @etf=?")
return pandas.read_sql_query(sql=query, con=connection, params=[etf]) …我有一个模型：
class Announcements(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(150), nullable=False)
text = db.Column(db.Text(), nullable=False)
category = db.Column(db.Integer(), nullable=False)
subcategory = db.Column(db.Integer(), nullable=False)
offer_type = db.Column(db.Integer(), nullable=False)
url = db.Column(db.String(150), nullable=True)
status = db.Column(db.Integer(), nullable=False)
def __repr__(self):
return '<Announcement %r>' % self.name
和公共功能：
def update(id):
// UPDATE HERE
如何使用 Flask-SQLAlchemy通过传入id的函数更新注释update()？
我在文档中没有看到任何更新行的示例
在我问之前，我会承认这是一个奇怪的问题。我想知道是否可以使用原始 SQL 而不是使用方法本身来复制 Flask/SQL Alchemy 类方法？
长话短说，我和我的队友正在参加数据库设计课程，我们现在处于实施阶段，我们正在编写基于我们的数据库架构设计的应用程序。我们想让事情变得简单，所以我们选择在 Python 中使用 Flask。我们正在关注Flask Mega 教程，这是一个解释如何像我们所做的那样构建基本站点的精彩教程。我们刚刚完成了第 5 章：用户登录，并且正在继续。
在app/routes.py脚本中，教程做了一些事情来获取用户信息。以下是示例应用程序的示例登录路径：
from flask_login import current_user, login_user
from app.models import User
# ...
@app.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('index'))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user is None or not user.check_password(form.password.data):
flash('Invalid username or password')
return redirect(url_for('login'))
login_user(user, remember=form.remember_me.data)
return redirect(url_for('index'))
return render_template('login.html', title='Sign In', form=form)
该行user = User.query.filter_by(username=form.username.data).first()是我感兴趣的。基本上，该行实例化了User类，它是来自 SQL …