test_alchemy

import asyncio
import datetime
from time import time

from sqlalchemy import Column, Integer, DateTime, String, func
from sqlalchemy.orm import sessionmaker
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from faker import Faker
from faker.providers import date_time
from databases import Database

Base = declarative_base()


class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    date = Column(DateTime, server_default=func.now())


a = A.__table__


async def main():
    db_url = 'postgresql+asyncpg://postgres:nipomu54@localhost:5432/testalch'
    engine = create_async_engine(db_url,
                                 convert_unicode=True)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async_session = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

    fake = Faker()
    fake.add_provider(date_time)
    test_data = [A(name=fake.name(), date=fake.date_time()) for _ in range(1000)]

    async with async_session() as session:
        async with session.begin():
            session.add_all(test_data)

        query = select(A).where(A.date > datetime.datetime(2008, 1, 1))
        res = []
        for i in range(100):
            start_time = time()
            await session.execute(query)
            res.append(time() - start_time)
        print(sum(res)/len(res))
        await session.commit()

    db = Database(db_url)
    await db.connect()
    query = a.select().where(a.c.date > datetime.datetime(2008, 1, 1))
    res = []
    for i in range(100):
        start_time = time()
        await db.fetch_all(query)
        res.append(time() - start_time)
    print(sum(res)/len(res))
    await db.disconnect()


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