alex
0d0f41d616
- Full CRUD endpoints for backup sources - Pydantic validation for source types - pytest tests for create, list, get, delete
62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from typing import List
|
|
from app.database import get_db
|
|
from app.models import Source
|
|
from app.schemas import SourceCreate, SourceUpdate, Source as SourceSchema
|
|
|
|
router = APIRouter(prefix="/api/sources", tags=["sources"])
|
|
|
|
@router.get("/", response_model=List[SourceSchema])
|
|
async def list_sources(db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(select(Source))
|
|
sources = result.scalars().all()
|
|
return sources
|
|
|
|
@router.post("/", response_model=SourceSchema)
|
|
async def create_source(source: SourceCreate, db: AsyncSession = Depends(get_db)):
|
|
db_source = Source(**source.model_dump())
|
|
db.add(db_source)
|
|
await db.commit()
|
|
await db.refresh(db_source)
|
|
return db_source
|
|
|
|
@router.get("/{source_id}", response_model=SourceSchema)
|
|
async def get_source(source_id: int, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(select(Source).where(Source.id == source_id))
|
|
source = result.scalar_one_or_none()
|
|
if not source:
|
|
raise HTTPException(status_code=404, detail="Source not found")
|
|
return source
|
|
|
|
@router.put("/{source_id}", response_model=SourceSchema)
|
|
async def update_source(
|
|
source_id: int,
|
|
source_update: SourceUpdate,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
result = await db.execute(select(Source).where(Source.id == source_id))
|
|
source = result.scalar_one_or_none()
|
|
if not source:
|
|
raise HTTPException(status_code=404, detail="Source not found")
|
|
|
|
update_data = source_update.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(source, field, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(source)
|
|
return source
|
|
|
|
@router.delete("/{source_id}")
|
|
async def delete_source(source_id: int, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(select(Source).where(Source.id == source_id))
|
|
source = result.scalar_one_or_none()
|
|
if not source:
|
|
raise HTTPException(status_code=404, detail="Source not found")
|
|
|
|
await db.delete(source)
|
|
await db.commit()
|
|
return {"message": "Source deleted"}
|