时间戳

首页 > 解决方案 > 使用 SQLAlchemy Session 作为上下文管理器时收到“AttributeError：__enter__”

python - 使用 SQLAlchemy Session 作为上下文管理器时收到“AttributeError：__enter__”

问题描述

我收到AttributeError: __enter__以下内容。这与with Session(engine) as session：

from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.orm import Session
from sqlalchemy import MetaData
from sqlalchemy import Table, Column, Integer, String
from sqlalchemy import ForeignKey

engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)

with engine.connect() as conn:
    conn.execute(text("CREATE TABLE some_table (x int, y int)"))
    conn.execute(text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),[{"x": 1, "y": 1}, {"x": 2, "y": 4}])

with engine.begin() as conn:
    conn.execute(text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),[{"x": 6, "y": 8}, {"x": 9, "y": 10},
    {"x": 11, "y": 12}, {"x": 13, "y": 14}])

with engine.connect() as conn:
    result = conn.execute(text("Select x,y From some_table"))
    for x, y in result:
        print(f"x:{x} y:{y}")

stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y").bindparams(y=6)

with Session(engine) as session:
    result = session.execute(stmt)
    for row in result:
        print(f'x: {row.x}  y: {row.y}')

我正在使用 Anaconda 1.3.23 中包含的 SQLAlchemy 版本。

标签： pythonsqlalchemy

解决方案

通过上下文管理器运行会话构建/关闭过程，如下所示：

engine = create_engine(...)
Session = sessionmaker(bind=engine)

with Session() as session:
    session.add(something)
    session.commit()

SQLAlchemy < 不支持1.4。

如果您的 SQLAlchemy 版本是 eg 1.3.x，您应该改为：

engine = create_engine(...)
Session = sessionmaker(bind=engine)

session = Session()
session.add(something)
session.commit()

如果你真的想使用上下文管理器，同时你需要使用 SQLAlchemy < 1.4，你可以使用以下方法（从SQLAlchemy docs复制）：

### another way (but again *not the only way*) to do it ###

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)

推荐阅读