SqlAlchemy 2.0 中文文档（二十八）（1）

attribute sqlalchemy.orm.exc.ConcurrentModificationError

exception sqlalchemy.orm.exc.DetachedInstanceError

exception sqlalchemy.orm.exc.FlushError

exception sqlalchemy.orm.exc.LoaderStrategyException

method __init__(applied_to_property_type: Type[Any], requesting_property: MapperProperty[Any], applies_to: Type[MapperProperty[Any]] | None, actual_strategy_type: Type[LoaderStrategy] | None, strategy_key: Tuple[Any, ...])

sqlalchemy.orm.exc.NO_STATE = (<class 'AttributeError'>, <class 'KeyError'>)

exception sqlalchemy.orm.exc.ObjectDeletedError

method __init__(state: InstanceState[Any], msg: str | None = None)

exception sqlalchemy.orm.exc.ObjectDereferencedError

exception sqlalchemy.orm.exc.StaleDataError

exception sqlalchemy.orm.exc.UnmappedClassError

method __init__(cls: Type[_T], msg: str | None = None)

exception sqlalchemy.orm.exc.UnmappedColumnError

exception sqlalchemy.orm.exc.UnmappedError

exception sqlalchemy.orm.exc.UnmappedInstanceError

method __init__(obj: object, msg: str | None = None)

x86_64 aarch64 ppc64le amd64 win32

pip install sqlalchemy[asyncio]

>>> import asyncio
>>> from sqlalchemy import Column
>>> from sqlalchemy import MetaData
>>> from sqlalchemy import select
>>> from sqlalchemy import String
>>> from sqlalchemy import Table
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> meta = MetaData()
>>> t1 = Table("t1", meta, Column("name", String(50), primary_key=True))
>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(meta.drop_all)
...         await conn.run_sync(meta.create_all)
...
...         await conn.execute(
...             t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
...         )
...
...     async with engine.connect() as conn:
...         # select a Result, which will be delivered with buffered
...         # results
...         result = await conn.execute(select(t1).where(t1.c.name == "some name 1"))
...
...         print(result.fetchall())
...
...     # for AsyncEngine created in function scope, close and
...     # clean-up pooled connections
...     await engine.dispose()
>>> asyncio.run(async_main())
BEGIN  (implicit)
...
CREATE  TABLE  t1  (
  name  VARCHAR(50)  NOT  NULL,
  PRIMARY  KEY  (name)
)
...
INSERT  INTO  t1  (name)  VALUES  (?)
[...]  [('some name 1',),  ('some name 2',)]
COMMIT
BEGIN  (implicit)
SELECT  t1.name
FROM  t1
WHERE  t1.name  =  ?
[...]  ('some name 1',)
[('some name 1',)]
ROLLBACK 

async with engine.connect() as conn:
    async_result = await conn.stream(select(t1))
    async for row in async_result:
        print("row: %s" % (row,))

>>> from __future__ import annotations
>>> import asyncio
>>> import datetime
>>> from typing import List
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import func
>>> from sqlalchemy import select
>>> from sqlalchemy.ext.asyncio import AsyncAttrs
>>> from sqlalchemy.ext.asyncio import async_sessionmaker
>>> from sqlalchemy.ext.asyncio import AsyncSession
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship
>>> from sqlalchemy.orm import selectinload
>>> class Base(AsyncAttrs, DeclarativeBase):
...     pass
>>> class B(Base):
...     __tablename__ = "b"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
...     data: Mapped[str]
>>> class A(Base):
...     __tablename__ = "a"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     data: Mapped[str]
...     create_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
...     bs: Mapped[List[B]] = relationship()
>>> async def insert_objects(async_session: async_sessionmaker[AsyncSession]) -> None:
...     async with async_session() as session:
...         async with session.begin():
...             session.add_all(
...                 [
...                     A(bs=[B(data="b1"), B(data="b2")], data="a1"),
...                     A(bs=[], data="a2"),
...                     A(bs=[B(data="b3"), B(data="b4")], data="a3"),
...                 ]
...             )
>>> async def select_and_update_objects(
...     async_session: async_sessionmaker[AsyncSession],
... ) -> None:
...     async with async_session() as session:
...         stmt = select(A).order_by(A.id).options(selectinload(A.bs))
...
...         result = await session.execute(stmt)
...
...         for a in result.scalars():
...             print(a, a.data)
...             print(f"created at: {a.create_date}")
...             for b in a.bs:
...                 print(b, b.data)
...
...         result = await session.execute(select(A).order_by(A.id).limit(1))
...
...         a1 = result.scalars().one()
...
...         a1.data = "new data"
...
...         await session.commit()
...
...         # access attribute subsequent to commit; this is what
...         # expire_on_commit=False allows
...         print(a1.data)
...
...         # alternatively, AsyncAttrs may be used to access any attribute
...         # as an awaitable (new in 2.0.13)
...         for b1 in await a1.awaitable_attrs.bs:
...             print(b1, b1.data)
>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     # async_sessionmaker: a factory for new AsyncSession objects.
...     # expire_on_commit - don't expire objects after transaction commit
...     async_session = async_sessionmaker(engine, expire_on_commit=False)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(Base.metadata.create_all)
...
...     await insert_objects(async_session)
...     await select_and_update_objects(async_session)
...
...     # for AsyncEngine created in function scope, close and
...     # clean-up pooled connections
...     await engine.dispose()
>>> asyncio.run(async_main())
BEGIN  (implicit)
...
CREATE  TABLE  a  (
  id  INTEGER  NOT  NULL,
  data  VARCHAR  NOT  NULL,
  create_date  DATETIME  DEFAULT  (CURRENT_TIMESTAMP)  NOT  NULL,
  PRIMARY  KEY  (id)
)
...
CREATE  TABLE  b  (
  id  INTEGER  NOT  NULL,
  a_id  INTEGER  NOT  NULL,
  data  VARCHAR  NOT  NULL,
  PRIMARY  KEY  (id),
  FOREIGN  KEY(a_id)  REFERENCES  a  (id)
)
...
COMMIT
BEGIN  (implicit)
INSERT  INTO  a  (data)  VALUES  (?)  RETURNING  id,  create_date
[...]  ('a1',)
...
INSERT  INTO  b  (a_id,  data)  VALUES  (?,  ?)  RETURNING  id
[...]  (1,  'b2')
...
COMMIT
BEGIN  (implicit)
SELECT  a.id,  a.data,  a.create_date
FROM  a  ORDER  BY  a.id
[...]  ()
SELECT  b.a_id  AS  b_a_id,  b.id  AS  b_id,  b.data  AS  b_data
FROM  b
WHERE  b.a_id  IN  (?,  ?,  ?)
[...]  (1,  2,  3)
<A  object  at  ...>  a1
created  at:  ...
<B  object  at  ...>  b1
<B  object  at  ...>  b2
<A  object  at  ...>  a2
created  at:  ...
<A  object  at  ...>  a3
created  at:  ...
<B  object  at  ...>  b3
<B  object  at  ...>  b4
SELECT  a.id,  a.data,  a.create_date
FROM  a  ORDER  BY  a.id
LIMIT  ?  OFFSET  ?
[...]  (1,  0)
UPDATE  a  SET  data=?  WHERE  a.id  =  ?
[...]  ('new data',  1)
COMMIT
new  data
<B  object  at  ...>  b1
<B  object  at  ...>  b2 

from __future__ import annotations
from typing import List
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import relationship
class Base(AsyncAttrs, DeclarativeBase):
    pass
class A(Base):
    __tablename__ = "a"
    # ... rest of mapping ...
    bs: Mapped[List[B]] = relationship()
class B(Base):
    __tablename__ = "b"
    # ... rest of mapping ...

a1 = (await session.scalars(select(A))).one()
for b1 in await a1.awaitable_attrs.bs:
    print(b1)

stmt = select(A).options(selectinload(A.bs))

A(bs=[], data="a2")

# create AsyncSession with expire_on_commit=False
async_session = AsyncSession(engine, expire_on_commit=False)
# sessionmaker version
async_session = async_sessionmaker(engine, expire_on_commit=False)
async with async_session() as session:
    result = await session.execute(select(A).order_by(A.id))
    a1 = result.scalars().first()
    # commit would normally expire all attributes
    await session.commit()
    # access attribute subsequent to commit; this is what
    # expire_on_commit=False allows
    print(a1.data)

# assume a_obj is an A that has lazy loaded A.bs collection
a_obj = await async_session.get(A, [1])
# force the collection to load by naming it in attribute_names
await async_session.refresh(a_obj, ["bs"])
# collection is present
print(f"bs collection: {a_obj.bs}")

user = await session.get(User, 42)
addresses = (await session.scalars(user.addresses.statement)).all()
stmt = user.addresses.statement.where(Address.email_address.startswith("patrick"))
addresses_filter = (await session.scalars(stmt)).all()

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
def fetch_and_update_objects(session):
  """run traditional sync-style ORM code in a function that will be
 invoked within an awaitable.
 """
    # the session object here is a traditional ORM Session.
    # all features are available here including legacy Query use.
    stmt = select(A)
    result = session.execute(stmt)
    for a1 in result.scalars():
        print(a1)
        # lazy loads
        for b1 in a1.bs:
            print(b1)
    # legacy Query use
    a1 = session.query(A).order_by(A.id).first()
    a1.data = "new data"
async def async_main():
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/test",
        echo=True,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncSession(engine) as session:
        async with session.begin():
            session.add_all(
                [
                    A(bs=[B(), B()], data="a1"),
                    A(bs=[B()], data="a2"),
                    A(bs=[B(), B()], data="a3"),
                ]
            )
        await session.run_sync(fetch_and_update_objects)
        await session.commit()
    # for AsyncEngine created in function scope, close and
    # clean-up pooled connections
    await engine.dispose()
asyncio.run(async_main())

import asyncio
from sqlalchemy import event
from sqlalchemy import text
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
# connect event on instance of Engine
@event.listens_for(engine.sync_engine, "connect")
def my_on_connect(dbapi_con, connection_record):
    print("New DBAPI connection:", dbapi_con)
    cursor = dbapi_con.cursor()
    # sync style API use for adapted DBAPI connection / cursor
    cursor.execute("select 'execute from event'")
    print(cursor.fetchone()[0])
# before_execute event on all Engine instances
@event.listens_for(Engine, "before_execute")
def my_before_execute(
    conn,
    clauseelement,
    multiparams,
    params,
    execution_options,
):
    print("before execute!")
async def go():
    async with engine.connect() as conn:
        await conn.execute(text("select 1"))
    await engine.dispose()
asyncio.run(go())

New DBAPI connection: <AdaptedConnection <asyncpg.connection.Connection object at 0x7f33f9b16960>>
execute from event
before execute!

import asyncio
from sqlalchemy import event
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import Session
engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost:5432/test")
session = AsyncSession(engine)
# before_commit event on instance of Session
@event.listens_for(session.sync_session, "before_commit")
def my_before_commit(session):
    print("before commit!")
    # sync style API use on Session
    connection = session.connection()
    # sync style API use on Connection
    result = connection.execute(text("select 'execute from event'"))
    print(result.first())
# after_commit event on all Session instances
@event.listens_for(Session, "after_commit")
def my_after_commit(session):
    print("after commit!")
async def go():
    await session.execute(text("select 1"))
    await session.commit()
    await session.close()
    await engine.dispose()
asyncio.run(go())

before commit!
execute from event
after commit!

import asyncio
from sqlalchemy import event
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import sessionmaker
sync_maker = sessionmaker()
maker = async_sessionmaker(sync_session_class=sync_maker)
@event.listens_for(sync_maker, "before_commit")
def before_commit(session):
    print("before commit")
async def main():
    async_session = maker()
    await async_session.commit()
asyncio.run(main())

before commit

from sqlalchemy import event
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(...)
@event.listens_for(engine.sync_engine, "connect")
def register_custom_types(dbapi_connection, *args):
    dbapi_connection.run_async(
        lambda connection: connection.set_type_codec(
            "MyCustomType",
            encoder,
            decoder,  # ...
        )
    )

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host/dbname",
    poolclass=NullPool,
)

from asyncio import current_task
from sqlalchemy.ext.asyncio import (
    async_scoped_session,
    async_sessionmaker,
)
async_session_factory = async_sessionmaker(
    some_async_engine,
    expire_on_commit=False,
)
AsyncScopedSession = async_scoped_session(
    async_session_factory,
    scopefunc=current_task,
)
some_async_session = AsyncScopedSession()

async def some_function(some_async_session, some_object):
    # use the AsyncSession directly
    some_async_session.add(some_object)
    # use the AsyncSession via the context-local proxy
    await AsyncScopedSession.commit()
    # "remove" the current proxied AsyncSession for the local context
    await AsyncScopedSession.remove()

import asyncio
from sqlalchemy import inspect
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test")
def use_inspector(conn):
    inspector = inspect(conn)
    # use the inspector
    print(inspector.get_view_names())
    # return any value to the caller
    return inspector.get_table_names()
async def async_main():
    async with engine.connect() as conn:
        tables = await conn.run_sync(use_inspector)
asyncio.run(async_main())

function sqlalchemy.ext.asyncio.create_async_engine(url: str | URL, **kw: Any) → AsyncEngine

function sqlalchemy.ext.asyncio.async_engine_from_config(configuration: Dict[str, Any], prefix: str = 'sqlalchemy.', **kwargs: Any) → AsyncEngine

function sqlalchemy.ext.asyncio.create_async_pool_from_url(url: str | URL, **kwargs: Any) → Pool

class sqlalchemy.ext.asyncio.AsyncEngine

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname")

method begin() → AsyncIterator[AsyncConnection]

async with async_engine.begin() as conn:
    await conn.execute(
        text("insert into table (x, y, z) values (1, 2, 3)")
    )
    await conn.execute(text("my_special_procedure(5)"))

method clear_compiled_cache() → None

method connect() → AsyncConnection

async with async_engine.connect() as conn:
    result = await conn.execute(select(user_table))

attribute dialect

method async dispose(close: bool = True) → None

attribute driver

attribute echo

attribute engine

method execution_options(**opt: Any) → AsyncEngine

method get_execution_options() → _ExecuteOptions

attribute name

attribute pool

method async raw_connection() → PoolProxiedConnection

attribute sync_engine: Engine

method update_execution_options(**opt: Any) → None

attribute url

class sqlalchemy.ext.asyncio.AsyncConnection

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname")
async with engine.connect() as conn:
    result = await conn.execute(select(table))

method async aclose() → None

method begin() → AsyncTransaction

method begin_nested() → AsyncTransaction

method async close() → None

attribute closed

method async commit() → None

attribute connection

attribute default_isolation_level

attribute dialect

method async exec_driver_sql(statement: str, parameters: _DBAPIAnyExecuteParams | None = None, execution_options: CoreExecuteOptionsParameter | None = None) → CursorResult[Any]

method async execute(statement: Executable, parameters: _CoreAnyExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → CursorResult[Any]

method async execution_options(**opt: Any) → AsyncConnection

method get_nested_transaction() → AsyncTransaction | None

method async get_raw_connection() → PoolProxiedConnection

method get_transaction() → AsyncTransaction | None

method in_nested_transaction() → bool

method in_transaction() → bool

attribute info

method async invalidate(exception: BaseException | None = None) → None

attribute invalidated

method async rollback() → None

method async run_sync(fn: ~typing.Callable[[~typing.Concatenate[~sqlalchemy.engine.base.Connection, ~_P]], ~sqlalchemy.ext.asyncio.engine._T], *arg: ~typing.~_P, **kw: ~typing.~_P) → _T

def do_something_with_core(conn: Connection, arg1: int, arg2: str) -> str:
  '''A synchronous function that does not require awaiting
 :param conn: a Core SQLAlchemy Connection, used synchronously
 :return: an optional return value is supported
 '''
    conn.execute(
        some_table.insert().values(int_col=arg1, str_col=arg2)
    )
    return "success"
async def do_something_async(async_engine: AsyncEngine) -> None:
  '''an async function that uses awaiting'''
    async with async_engine.begin() as async_conn:
        # run do_something_with_core() with a sync-style
        # Connection, proxied into an awaitable
        return_code = await async_conn.run_sync(do_something_with_core, 5, "strval")
        print(return_code)

# run metadata.create_all(conn) with a sync-style Connection,
# proxied into an awaitable
with async_engine.begin() as conn:
    await conn.run_sync(metadata.create_all)

method async scalar(statement: Executable, parameters: _CoreSingleExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → Any

method async scalars(statement: Executable, parameters: _CoreAnyExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → ScalarResult[Any]

method async start(is_ctxmanager: bool = False) → AsyncConnection

method stream(statement: Executable, parameters: _CoreAnyExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → AsyncIterator[AsyncResult[Any]]

result = await conn.stream(stmt):
async for row in result:
    print(f"{row}")

async with conn.stream(stmt) as result:
    async for row in result:
        print(f"{row}")

method stream_scalars(statement: Executable, parameters: _CoreSingleExecuteParams | None = None, *, execution_options: CoreExecuteOptionsParameter | None = None) → AsyncIterator[AsyncScalarResult[Any]]

result = await conn.stream_scalars(stmt)
async for scalar in result:
    print(f"{scalar}")

async with conn.stream_scalars(stmt) as result:
    async for scalar in result:
        print(f"{scalar}")

attribute sync_connection: Connection | None

attribute sync_engine: Engine

class sqlalchemy.ext.asyncio.AsyncTransaction

method async close() → None

method async commit() → None

method async rollback() → None

method async start(is_ctxmanager: bool = False) → AsyncTransaction