71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from pathlib import Path
|
|
from typing import List
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, Request, Response
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlmodel import Session, select
|
|
|
|
from mapmakr.models import Marker, MarkerRead, MarkerCreate, MarkerUpdate, create_db_tables, \
|
|
get_session
|
|
from mapmakr.settings import settings
|
|
|
|
|
|
ROOT = Path(__file__).parent
|
|
|
|
app = FastAPI()
|
|
app.mount('/static', StaticFiles(directory=str(ROOT / 'static')), name='static')
|
|
templates = Jinja2Templates(directory=str(ROOT / 'templates'))
|
|
|
|
|
|
@app.on_event('startup')
|
|
def on_startup():
|
|
create_db_tables()
|
|
|
|
|
|
@app.get('/', response_class=HTMLResponse)
|
|
async def index(request: Request):
|
|
return templates.TemplateResponse('index.html', {'request': request,
|
|
'site_title': settings.site_title})
|
|
|
|
|
|
@app.post('/marker', tags=['markers'], response_model=MarkerRead)
|
|
async def create_marker(marker: MarkerCreate, session: Session = Depends(get_session)) -> Marker:
|
|
db_marker = Marker.from_orm(marker)
|
|
session.add(db_marker)
|
|
session.commit()
|
|
session.refresh(db_marker)
|
|
return db_marker
|
|
|
|
|
|
@app.patch('/marker/{marker_id}', tags=['markers'], response_model=MarkerRead)
|
|
async def update_marker(marker_id: int, marker: MarkerUpdate,
|
|
session: Session = Depends(get_session)) -> Marker:
|
|
db_marker = session.get(Marker, marker_id)
|
|
if not db_marker:
|
|
raise HTTPException(status_code=404, detail='Marker not found')
|
|
marker_data = marker.dict(exclude_unset=True)
|
|
for key, value in marker_data.items():
|
|
setattr(db_marker, key, value)
|
|
session.add(db_marker)
|
|
session.commit()
|
|
session.refresh(db_marker)
|
|
return db_marker
|
|
|
|
|
|
@app.get('/marker', tags=['markers'], response_model=List[MarkerRead])
|
|
async def read_markers(session: Session = Depends(get_session)) -> list[MarkerRead]:
|
|
markers = session.exec(select(Marker)).all()
|
|
return markers
|
|
|
|
|
|
@app.delete('/marker/{marker_id}', tags=['markers'])
|
|
async def delete_marker(marker_id: int, session: Session = Depends(get_session)):
|
|
db_marker = session.get(Marker, marker_id)
|
|
if not db_marker:
|
|
raise HTTPException(status_code=404, detail='Marker not found')
|
|
session.delete(db_marker)
|
|
session.commit()
|
|
return Response(status_code=200)
|