90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
import uuid
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, Depends, HTTPException, Query, Header
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import get_db, engine, Base
|
|
from app.models import Paste
|
|
from app.schemas import (
|
|
PasteCreate,
|
|
PasteResponse,
|
|
PasteCreatedResponse,
|
|
PasteListResponse,
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="Pastebin API", lifespan=lifespan)
|
|
|
|
|
|
def verify_admin(password: str = Header(..., alias="X-Admin-Password")):
|
|
if password != settings.admin_password:
|
|
raise HTTPException(status_code=401, detail="Invalid admin password")
|
|
|
|
|
|
@app.post("/api/pastes", response_model=PasteCreatedResponse, status_code=201)
|
|
async def create_paste(data: PasteCreate, db: AsyncSession = Depends(get_db)):
|
|
paste = Paste(content=data.content)
|
|
db.add(paste)
|
|
await db.commit()
|
|
await db.refresh(paste)
|
|
return PasteCreatedResponse(
|
|
id=paste.id, delete_token=paste.delete_token, created_at=paste.created_at
|
|
)
|
|
|
|
|
|
@app.get("/api/pastes/{paste_id}", response_model=PasteResponse)
|
|
async def get_paste(paste_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
|
|
paste = await db.get(Paste, paste_id)
|
|
if not paste:
|
|
raise HTTPException(status_code=404, detail="Paste not found")
|
|
return PasteResponse(id=paste.id, content=paste.content, created_at=paste.created_at)
|
|
|
|
|
|
@app.delete("/api/pastes/{paste_id}")
|
|
async def delete_paste(
|
|
paste_id: uuid.UUID,
|
|
token: uuid.UUID = Query(..., alias="token"),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
paste = await db.get(Paste, paste_id)
|
|
if not paste:
|
|
raise HTTPException(status_code=404, detail="Paste not found")
|
|
if paste.delete_token != token:
|
|
raise HTTPException(status_code=403, detail="Invalid delete token")
|
|
await db.delete(paste)
|
|
await db.commit()
|
|
return {"detail": "Paste deleted"}
|
|
|
|
|
|
@app.get("/api/admin/pastes", response_model=list[PasteListResponse])
|
|
async def list_pastes(
|
|
_=Depends(verify_admin),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(Paste).order_by(Paste.created_at.desc()))
|
|
return result.scalars().all()
|
|
|
|
|
|
@app.delete("/api/admin/pastes/{paste_id}")
|
|
async def admin_delete_paste(
|
|
paste_id: uuid.UUID,
|
|
_=Depends(verify_admin),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
paste = await db.get(Paste, paste_id)
|
|
if not paste:
|
|
raise HTTPException(status_code=404, detail="Paste not found")
|
|
await db.delete(paste)
|
|
await db.commit()
|
|
return {"detail": "Paste deleted"}
|