Код: Выделить всё
uvicorn main:app --reload
Вот мой код на данный момент:
Код: Выделить всё
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Annotated
from sqlalchemy.orm import Session
from sqlalchemy import Column, Integer, String, text
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from contextlib import asynccontextmanager
DATABASE_URL = "postgresql+asyncpg://$DB_USER:$DB_PWD@$DB_SERVER/$DB_NAME"
Base = declarative_base()
class Person(Base):
__tablename__ = "persons"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
age = Column(Integer)
engine = create_async_engine(DATABASE_URL, future=True, echo=True)
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
app = FastAPI()
@asynccontextmanager
async def lifespan(app: FastAPI):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
await seed_database(conn)
yield
await engine.dispose()
app.lifespan_context = lifespan
async def seed_database(conn):
await conn.execute(text("INSERT INTO persons (name, age) VALUES ('Alice', 23)"))
await conn.execute(text("INSERT INTO persons (name, age) VALUES ('Bob', 22)"))
await conn.commit()
print("Database seeded with initial persons.")
@app.get("/")
async def read_root(db: AsyncSession = Depends(lambda: SessionLocal())):
result = await db.execute(text("SELECT * FROM persons"))
persons = result.mappings().all()
return {"persons": persons}
Подробнее здесь: https://stackoverflow.com/questions/793 ... nc-fastapi
Мобильная версия