标签: sqlalchemy

我刚刚开始使用SQLAlchemy,但仍然无法解决一些概念问题.

归结为基本元素,我有两个这样的表(这是通过Flask-SQLAlchemy):

class User(db.Model):
    __tablename__ = 'users'
    user_id = db.Column(db.Integer, primary_key=True)

class Posts(db.Model):
    __tablename__ = 'posts'
    post_id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.user_id'))
    post_time = db.Column(db.DateTime)

    user = db.relationship('User', backref='posts')

我将如何查询用户列表及其最新帖子(不包括没有帖子的用户).如果我使用SQL,我会这样做:

SELECT [whatever]
FROM posts AS p
    LEFT JOIN users AS u ON u.user_id = p.user_id
WHERE p.post_time = (SELECT MAX(post_time) FROM posts WHERE user_id = u.user_id)

所以我确切地知道"想要的"SQL以获得我想要的效果,但不知道如何在SQLAlchemy中"正确地"表达它.

编辑:如果它很重要,我在SQLAlchemy 0.6.6上.

我有一个问题:

MyModel.query.filter(Mymodel.name.contains('a_string'))

我需要做同样的查询但是使用否定(不喜欢运算符)但是在SQLAlchemy文档中没有找到符合我需要的运算符.

有没有办法不使用SQLAlchemy的SQL部分???

我知道这可能是一个简单的问题,但我不知道解决方案.当我尝试连接到postgresql时,这里发生了什么？我是这个数据库和编程领域的自学者,所以请温柔地对待我.当我尝试以下代码时:

import sqlalchemy
db = sqlalchemy.create_engine('postgresql:///tutorial.db')

我收到此错误:

回溯(最近一次调用最后一次):文件"",第1行,db = sqlalchemy.create_engine('postgresql:///tutorial.db')文件"C:\ Python27\lib\site-packages\sqlalchemy-0.7. 5dev-py2.7.egg\sqlalchemy\engine__init __.py",第327行,在create_engine中返回strategy.create(*args,**kwargs)文件"C:\ Python27\lib\site-packages\sqlalchemy-0.7.5dev -py2.7.egg\sqlalchemy\engine\strategies.py",第64行,在create dbapi = dialect_cls.dbapi(**dbapi_args)文件"C:\ Python27\lib\site-packages\sqlalchemy-0.7.5dev- py2.7.egg\sqlalchemy\dialects\postgresql\psycopg2.py",第289行,dbapi psycopg = import('psycopg2')ImportError:没有名为psycopg2的模块

我需要单独安装psycopg2吗？postgresql的正确连接字符串是什么？

有没有人有经验分析Python/SQLAlchemy应用程序？找到瓶颈和设计缺陷的最佳方法是什么？

我们有一个Python应用程序,其中数据库层由SQLAlchemy处理.该应用程序使用批处理设计,因此许多数据库请求按顺序在有限的时间内完成.目前运行时间太长,因此需要进行一些优化.我们不使用ORM功能,数据库是PostgreSQL.

使用SQLAlchemy 0.5.8如何运行最大查询？

即 Select max(id) from some_table

我正在开发一个基于exisitng数据库的Pylons应用程序,所以我正在使用反射.我有一个SQL文件,其中包含我用来创建测试数据库的模式.这就是为什么我不能简单地使用drop_all和create_all.

我想写一些单元测试,我遇到了每次测试后清除数据库内容的问题.我只想删除所有数据,但保持表格完好无损.这可能吗？

该应用程序使用Postgres,这也是必须用于测试的内容.

我正在使用 async_engine。当我尝试执行任何操作时：

async with self.async_engine.connect() as con:
    query = "SELECT id, name FROM item LIMIT 50;"
    result = await con.execute(f"{query}")

我越来越：

Exception has occurred: ObjectNotExecutableError
Not an executable object: 'SELECT id, name FROM item LIMIT 50;'

这个问题之前由用户@stilmaniac提出过，但现在已从SO 中删除。

我在谷歌搜索缓存中找到了它，这里是副本。

我有同样的问题，所以我重新询问它，但原始版本如下：

我正在尝试从元数据创建表，如下所示：

Base = declarative_base()

properties = Table(
    'properties', Base.metadata,
    # ...
    Column('geolocation', Geography(geometry_type='POINT', srid=4326)),
    # ... 
)

engine = create_async_engine("postgresql+asyncpg://user:password@postgres/")
async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

给我以下错误：

sqlalchemy.exc.ObjectNotExecutableError: Not an executable object: 'CREATE INDEX …

使用声明式在SQLAlchemy中创建一对一关系的最佳方法是什么？

我有两个表,foo而且bar,我想foo.bar_id链接到bar.问题在于这是一种单向一对一的关系.bar一定不能知道foo.对于每一个foo,都会有一个且只有一个bar.

理想情况下,在选择foo后,我可以这样做:

myfoo.bar.whatever = 5 

使用声明式实现此目的的最佳方法是什么？

使用SQLAlchemy时,缓解SQL注入攻击的最佳做法是什么？

我在mysql db表中有一个布尔字段.

# table model
class TestCase(Base):
    __tablename__ = 'test_cases'
    ...
    obsoleted = Column('obsoleted',  Boolean)

要获得所有未经过时的测试用例的计数,可以像这样完成:

caseNum = session.query(TestCase).filter(TestCase.obsoleted == False).count()
print(caseNum)

这很好,但是flake8报告了以下警告:

E712:与False的比较应该是"如果cond是假的:"或"如果不是cond:"

好的,我觉得这很有道理.所以将我的代码更改为:

caseNum = session.query(TestCase).filter(TestCase.obsoleted is False).count()

要么

caseNum = session.query(TestCase).filter(not TestCase.obsoleted).count()

但他们都不能奏效.结果始终为0.我认为filter子句不支持运算符"is"或"is not".有人可以告诉我如何处理这种情况.我不想禁用片状.