Async SQLAlchemy cannot create tables in the database

Question

main.py

from fastapi import FastAPI
import uvicorn
from database.connection import init_models

app = FastAPI()

@app.on_event('startup')
async def init_db() -> None:
    await init_models()

if __name__ == '__main__':
    uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True)

connection.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from dotenv import load_dotenv
import os

load_dotenv()

DATABASE_URL: str = f'postgresql+asyncpg://{os.getenv("POSTGRES_USER")}:{os.getenv("POSTGRES_PASSWORD")}@localhost:5432/postgres'
SECRET_KEY: str = os.getenv('SECRET_KEY')

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()

async def init_models() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

I get such an output:

INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [4529] using StatReload
INFO:     Started server process [4534]
INFO:     Waiting for application startup.
2023-03-14 13:48:57,835 INFO sqlalchemy.engine.Engine select pg_catalog.version()
2023-03-14 13:48:57,835 INFO sqlalchemy.engine.Engine [raw sql] ()
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine select current_schema()
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine [raw sql] ()
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine show standard_conforming_strings
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine [raw sql] ()
2023-03-14 13:48:57,837 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-03-14 13:48:57,837 INFO sqlalchemy.engine.Engine COMMIT
INFO:     Application startup complete.

SQLAlchemy does not create any table. I have tried to use an usual session not async, tried to use alembic, nothing helps. There is everything ok with models, I cannot understand with which is the problem related.

Answer 1

You can use this code:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.engine.url import URL
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, String
from uuid import uuid4
from asyncio import run

# ======================================================== Connect to db
postgres_config = {"drivername": "postgresql+asyncpg",
                   "host": "127.0.0.1",
                   "port": "5432",
                   "database": "test_db",
                   "username": "root",
                   "password": "pass"}

Postgres_URL = URL.create(**postgres_config)

engine = create_async_engine(Postgres_URL,
                             echo=False,
                             pool_size=1000)

async_session = sessionmaker(engine,
                             autocommit=False,
                             future=True,
                             expire_on_commit=False,
                             class_=AsyncSession)

# ======================================================== Create Model
Base = declarative_base()


class db_model_countrys(Base):
    __tablename__ = "countriess"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    country_name = Column(String(15), nullable=False)
    country_english_name = Column(String(15), nullable=False)
    nationality = Column(String(15))
    abbreviation = Column(String(10))
    phone_code = Column(String(5))


# ======================================================== Create Model
async def create_tables() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# ======================================================== run
if __name__ == "__main__":
    run(create_tables())