首页 > 代码库 > SQLAlchemy ORM 参考

SQLAlchemy ORM 参考

2024-08-07 09:50:30   221人阅读

ORM

Session

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(‘sqlite:///:memory:‘, echo=False)
Session = sessionmaker(bind=engine)
session = Session()

echo 负责设定是否输出生成的 SQL 语句。

create_engine 返回的是一个 Engine 对象，它负责对接 DBAPI。Session 对象会更多被直接使用。

session 某种意义上类似于一个缓存，他有 commit, rollback, close 等方法。也因此 session 实例的生命周期维护应该位于使用它们的函数之外，与具体的应用逻辑区分开，以避免数据污染和保证 rollback/close 的调用。如：– 参考

from contextlib import contextmanager

@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def run_my_program():
    with session_scope() as session:
        ThingOne().go(session)
        ThingTwo().go(session)

映射对象的声明

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = ‘user‘

    id = Column(Integer, primary_key=True)
    name = Column(String)

从 Python 类到 sql 表，需要一套映射逻辑，sqlalchemy 提供了一套默认的解决方案，就是 declarative_base。

这里单独的一行 Base=declarative_base() 表示了你可以有很多种方式自定义基类，比如继承 Base，或者给 declarative_base(**kwargs) 传参。– 参考

至于 User 类，由于 sqlalchemy 默认不会去做任何推导，你需要显示指定 __tablename__ 参数和至少一个主键。

Integer 类型没有指定长度，是因为这种修饰参数仅在建表时有用。

Query

new

tim = User(name=‘tim‘)
session.add(tim)
session.flush()

sqlalchemy 对待请求的处理方式是惰性的，他只在必要的时候才访问数据库。必要的一种定义是 query：

session.query(User).filter_by(name=‘tim‘)

立即同步到数据库的方式是使用 .flush() 方法。

另外一个非常有趣的特性是，sqlalchemy 能够在同一个 session 中分别出表记录的唯一性。即：

>>> tim is session.query(User).filter_by(name=‘tim‘)
>>> True

对于同一条数据库记录，不管是你新建的，还是多次从库中取出的，他们的 Python id 都是一样的。这种区分就是前面提到的 映射声明必须提供主键 所带来的一个好处。

query 对象

query() 的参数可以很多变，可以是一个 Model 类，或他的某些属性的组合：

foo = session.query(User, User.id).filter(User.name==‘tim‘).first()

这种 query 方式返回的元组对象都是关键字命名的，而且支持属性式访问，如：

>>> foo.User.id == foo.id
>>> True

如果想起别名，可以使用 .label() 方法：

>>> foo = session.query(User.id.label(‘user_id‘)).filter_by(name=‘tim‘)
>>> foo.user_id
>>> 1

对于 LIMIT 和 OFFSET 参数，可以直接对 Query 对象使用 Python 切片操作。

session.query(User)[20:10]

最后，Query 对象是可迭代的。尽管上面的例子为了偷懒没有使用过迭代方法，但它是支持的！

filter

filter() 较于 filter_by() 是一种更通用的方法，支持所有的 sql 表达式。filter() 返回的仍然是一个 Query 对象，所以你可以连续调用 filter() 方法，或者图省事把复合查询语句放进同一个 filter() 里

session.query(User).filter(id>1, name==‘tim‘)

session.query(User).filter(id>1).filter(name==‘tim‘)

一些常用的表达式语句：

  • User.name == ‘tim‘
  • User.name != ‘tim‘
  • User.name.like(‘tim‘)
  • User.name.in_([‘tim‘, ‘White‘])
  • ~User.name.in_([‘tim‘, ‘white‘])
  • User.name == None or User.name.is_(None)
  • User.name != None or User.name.isnot(None)
  • from sqlalchemy import or_ /// filter(or_(expr1, expr2))
  • User.name.match(‘tim‘)

all(), one(), first()

Query 对象虽然是可迭代的，但对于多数确定需求的状况，我们可能更希望直接把需要的数据取出来。

all() 和 first() 顾名思义，特别的是 one() 。

它较于 first() 多了一条约束，即Query 对象的结果集必须只包含一个对象，否则报错，即：

session.query(User).filter(name==‘tim‘).one()

等价于：

foo = session.quert(User).filter(name=‘tim‘)
assert len(foo.all()) == 1
return foo.first()

SQL

在如 filter() 或 order_by() 这样的方法中，条件表达式还可以是原始的 sql 语句，只需要用 text() 把语句包起来：

from sqlalchemy import text

users = session.query(User).filter(text(‘id<10‘)).order_by(text(‘id‘))

另一种更彻底的 sql 执行方式是使用 from_statement() 方法：

session.query(User).from_statement(text(‘SELECT * FROM user where id<10 and status=1‘)).all()

此时还可以直接 Query 列名：

>>> session.query(‘id‘, ‘name‘, ‘status‘).from_statement(‘SELECT * from user where class=3‘).all()
>>> [(1, ‘jack‘, 0), (2, ‘rose‘, 1)]

counting

一种简单的计数是对 Query 对象使用 count() 方法，用于返回 Query 对象的行数。

另一种较复杂的是 func.count()，它支持细分计数：

from sqlalchemy import func

session.query(func.count(User.class), User.class).filter_by(User.class=‘a‘)

session.query(func.count(User.class), User.class).group_by(User.class).all()

关系

略

SQLAlchemy ORM 参考


联系
我们