使用 SQLAlchemy 进行异步操作

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, create_engine, Column, Integer, String, DateTime, func, ForeignKey
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(120), unique=True)
    created_at = Column(DateTime, default=func.now())

    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}', created_at='{self.created_at}')>"

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(50))
    content = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=func.now())

    user = relationship("User", back_populates="posts")

    def __repr__(self):
        return f"<Post(title='{self.title}', content='{self.content}', user={self.user}, created_at='{self.created_at}')>"

User.posts = relationship("Post", order_by=Post.id, back_populates="user")

async def main():
    # create async engine using asyncpg driver
    POSTGRES_DB = "mydatabase"
    POSTGRES_USER = "myuser"
    POSTGRES_PASSWORD = "mypassword"
    POSTGRES_HOST = "localhost"
    POSTGRES_PORT = "5432"

    DATABASE_URI = f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"

    engine = create_async_engine(DATABASE_URI, future=True, echo=True)

    # create tables
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    # create async session
    async with AsyncSession(engine) as session:
        # create a user
        new_user = User(name="pidancode.com", email="pidancode@example.com")
        session.add(new_user)
        await session.commit()

        # query all users
        stmt = select(User)
        results = await session.execute(stmt)

        for row in results:
            print(row)

if __name__ == '__main__':
    asyncio.run(main())