89 lines
3 KiB
Django/Jinja
89 lines
3 KiB
Django/Jinja
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel import Session, select
|
|
|
|
from learn_sql_model.api.websocket_connection_manager import manager
|
|
from learn_sql_model.config import get_session
|
|
from learn_sql_model.models.{{ modelname }} import {{ modelname }}, {{ modelname }}Create, {{ modelname }}Read, {{ modelname }}Update, {{ modelname }}s
|
|
|
|
{{ modelname }}_router = APIRouter()
|
|
|
|
|
|
@{{ modelname }}_router.on_event("startup")
|
|
def on_startup() -> None:
|
|
# SQLModel.metadata.create_all(get_config().database.engine)
|
|
...
|
|
|
|
|
|
@{{ modelname }}_router.get("/{{ modelname }}/{{{ modelname }}_id}")
|
|
def get_{{ modelname }}(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
{{ modelname }}_id: int,
|
|
) -> {{ modelname }}Read:
|
|
"get one {{ modelname }}"
|
|
{{ modelname }} = session.get({{ modelname }}, {{ modelname }}_id)
|
|
if not {{ modelname }}:
|
|
raise HTTPException(status_code=404, detail="{{ modelname }} not found")
|
|
return {{ modelname }}
|
|
|
|
|
|
@{{ modelname }}_router.post("/{{ modelname }}/")
|
|
def post_{{ modelname }}(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
{{ modelname }}: {{ modelname }}Create,
|
|
) -> {{ modelname }}Read:
|
|
"create a {{ modelname }}"
|
|
db_{{ modelname }} = {{ modelname }}.from_orm({{ modelname }})
|
|
session.add(db_{{ modelname }})
|
|
session.commit()
|
|
session.refresh(db_{{ modelname }})
|
|
await manager.broadcast({{{ modelname }}.json()}, id=1)
|
|
return db_{{ modelname }}
|
|
|
|
|
|
@{{ modelname }}_router.patch("/{{ modelname }}/")
|
|
def patch_{{ modelname }}(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
{{ modelname }}: {{ modelname }}Update,
|
|
) -> {{ modelname }}Read:
|
|
"update a {{ modelname }}"
|
|
db_{{ modelname }} = session.get({{ modelname }}, {{ modelname }}.id)
|
|
if not db_{{ modelname }}:
|
|
raise HTTPException(status_code=404, detail="{{ modelname }} not found")
|
|
for key, value in {{ modelname }}.dict(exclude_unset=True).items():
|
|
setattr(db_{{ modelname }}, key, value)
|
|
session.add(db_{{ modelname }})
|
|
session.commit()
|
|
session.refresh(db_{{ modelname }})
|
|
await manager.broadcast({{{ modelname }}.json()}, id=1)
|
|
return db_{{ modelname }}
|
|
|
|
|
|
@{{ modelname }}_router.delete("/{{ modelname }}/{{{ modelname }}_id}")
|
|
def delete_{{ modelname }}(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
{{ modelname }}_id: int,
|
|
):
|
|
"delete a {{ modelname }}"
|
|
{{ modelname }} = session.get({{ modelname }}, {{ modelname }}_id)
|
|
if not {{ modelname }}:
|
|
raise HTTPException(status_code=404, detail="{{ modelname }} not found")
|
|
session.delete({{ modelname }})
|
|
session.commit()
|
|
await manager.broadcast(f"deleted {{ modelname }} {{{ modelname }}_id}", id=1)
|
|
return {"ok": True}
|
|
|
|
|
|
@{{ modelname }}_router.get("/{{ modelname }}s/")
|
|
def get_{{ modelname }}s(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
) -> {{ modelname }}s:
|
|
"get all {{ modelname }}s"
|
|
statement = select({{ modelname }})
|
|
{{ modelname }}s = session.exec(statement).all()
|
|
return {{ modelname }}s(__root__={{ modelname }}s)
|
|
|