</>
pythonadvanced

SQLAlchemy Async Session with asyncpg

Use SQLAlchemy 2.0 async sessions with asyncpg for non-blocking database access in async pipelines.

python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from sqlalchemy import Column, Integer, String, select
import asyncio

engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/db', echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = 'users'
    id   = Column(Integer, primary_key=True)
    name = Column(String)

async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User).limit(10))
        return result.scalars().all()

asyncio.run(get_users())

Use Cases

  • async web services
  • non-blocking DB access
  • FastAPI backends

Tags

#sqlalchemy#async#asyncpg#postgres

Related Snippets

Similar patterns you can reuse in the same workflow.

pythonintermediate

SQLAlchemy Bulk Insert with Upsert

Efficiently bulk-insert rows with conflict resolution using SQLAlchemy Core and PostgreSQL.

Best for: idempotent loads

#sqlalchemy#postgres
sqlbeginner

PostgreSQL COPY — Fast CSV Import

Use PostgreSQL COPY command for high-speed bulk data loading from CSV files with error handling.

Best for: High-speed bulk data loading into PostgreSQL

#postgres#copy
bashbeginner

Database Backup and Restore to S3

Automated PostgreSQL backup script with compression, S3 upload, retention policy, and restore commands.

Best for: Automated daily database backups to S3

#bash#backup
pythonintermediate

Async ETL Pipeline with asyncio

Run concurrent data fetches and transformations using asyncio.gather for high-throughput pipelines.

Best for: concurrent API ingestion

#asyncio#async