74 lines
2.9 KiB
Python
74 lines
2.9 KiB
Python
from fastapi import FastAPI, HTTPException, Depends
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import sessionmaker
|
|
from models import Auditory, Oboruds, Zametki, Base
|
|
from schemas import AuditoryCreate, Auditory, OborudsCreate, Oboruds, ZametkiCreate, Zametki # Импортируем Pydantic-модели
|
|
|
|
DATABASE_URL = "sqlite+aiosqlite:///./test.db" # Используем асинхронный SQLite
|
|
|
|
engine = create_async_engine(DATABASE_URL, echo=True)
|
|
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
|
|
|
|
app = FastAPI()
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
# Dependency to get the database session
|
|
async def get_db():
|
|
async with async_session() as session:
|
|
yield session
|
|
|
|
# CRUD для Auditory
|
|
@app.post("/auditory/", response_model=Auditory)
|
|
async def create_auditory(auditory: AuditoryCreate, db: AsyncSession = Depends(get_db)):
|
|
db_auditory = Auditory(**auditory.model_dump())
|
|
db.add(db_auditory)
|
|
await db.commit()
|
|
await db.refresh(db_auditory)
|
|
return db_auditory
|
|
|
|
@app.get("/auditory/{auditory_id}", response_model=Auditory)
|
|
async def read_auditory(auditory_id: int, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(select(Auditory).filter(Auditory.id == auditory_id))
|
|
auditory = result.scalars().first()
|
|
if auditory is None:
|
|
raise HTTPException(status_code=404, detail="Auditory not found")
|
|
return auditory
|
|
|
|
# CRUD для Oboruds
|
|
@app.post("/oboruds/", response_model=Oboruds)
|
|
async def create_oboruds(oboruds: OborudsCreate, db: AsyncSession = Depends(get_db)):
|
|
db_oboruds = Oboruds(**oboruds.model_dump())
|
|
db.add(db_oboruds)
|
|
await db.commit()
|
|
await db.refresh(db_oboruds)
|
|
return db_oboruds
|
|
|
|
@app.get("/oboruds/{oboruds_id}", response_model=Oboruds)
|
|
async def read_oboruds(oboruds_id: int, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(select(Oboruds).filter(Oboruds.id == oboruds_id))
|
|
oboruds = result.scalars().first()
|
|
if oboruds is None:
|
|
raise HTTPException(status_code=404, detail="Oboruds not found")
|
|
return oboruds
|
|
|
|
# CRUD для Zametki
|
|
@app.post("/zametki/", response_model=Zametki)
|
|
async def create_zametki(zametki: ZametkiCreate, db: AsyncSession = Depends(get_db)):
|
|
db_zametki = Zametki(**zametki.model_dump())
|
|
db.add(db_zametki)
|
|
await db.commit()
|
|
await db.refresh(db_zametki)
|
|
return db_zametki
|
|
|
|
@app.get("/zametki/{zametki_id}", response_model=Zametki)
|
|
async def read_zametki(zametki_id: int, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(select(Zametki).filter(Zametki.id == zametki_id))
|
|
zametki = result.scalars().first()
|
|
if zametki is None:
|
|
raise HTTPException(status_code=404, detail="Zametki not found")
|
|
return zametki |