with engine.begin() as connection: sql = 'SELECT COUNT(*) FROM samples' rows = connection.execute(text(sql)) count = list(rows)[0][0]
こんな方法ではなく、scalar() を使う
例)
from sqlalchemy import create_engine, text from sqlalchemy.orm import Session
# 接続情報、username, password, host, port, dbname は設定済 engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}') def countquery() -> int: with Session(autocommit=False, autoflush=True, bind=engine) as session: try: sql = 'SELECT COUNT(*) FROM samples' res = session.execute(text(sql)).scalar() except Exception as e: print(e) raise else: return res if __name__ == '__main__': c = countquery() print('count = %d' % c)
engine.begin() から実行する場合
with engine.begin() as connection: sql = 'SELECT COUNT(*) FROM samples' count = connection.execute(text(sql)).scalar()
Session query() で、DB の DUAL表を使う
from sqlalchemy import create_engine, text from sqlalchemy.orm import Session, registry import sqlalchemy as db
engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}') mapper_reg = registry() Base = mapper_reg.generate_base() class GenericNumeric(Base): __tablename__ = 'dual' count = db.Column(db.Numeric, nullable=False, primary_key=True) def countquery() -> int: with Session(autocommit=False, autoflush=True, bind=engine) as session: try: sql = 'SELECT COUNT(*) FROM samples' res = session.query(GenericNumeric).from_statement(text(sql)).scalar() except Exception as e: print(e) raise else: return res.count if __name__ == '__main__': c = countquery() print('count = %d' % c)
ちょっとナンセンスかな。。。