refactor to use pydantic settings and WAL sqlite mode

src/endpoints/default.py

@@ -6,20 +6,20 @@ from sqlalchemy.orm import Session
 
 from core.config import VERSION, BREAKING_CHANGES
 from db import crud
-from db.database import get_db
+from db.database import get_db_dependency, get_db
 from security import get_user_auth, bearer_security
 
-
 default_router = APIRouter()
 
+
 @default_router.get("/")
 async def home(request: Request):
     # TODO: maybe split into 2 routes: one non authenticated and one authenticated for the groups info only
     status = {"version": VERSION, "breakingChanges": BREAKING_CHANGES}
     try:
         email = await get_user_auth(await bearer_security(request))
-        db: Session = next(get_db())
-        status["groups"] = crud.get_user_groups(db, email)
+        with get_db() as db:
+            status["groups"] = crud.get_user_groups(db, email)
     except HTTPException: pass  # not authenticated is fine
     except Exception as e: logger.error(e)
     return JSONResponse(status)
@@ -29,8 +29,9 @@ async def home(request: Request):
 async def health():
     return JSONResponse({"status": "ok"})
 
+
 @default_router.get("/groups", response_model=list[str])
-def get_user_groups(db: Session = Depends(get_db), email=Depends(get_user_auth)):
+def get_user_groups(db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)):
     return crud.get_user_groups(db, email)
 
 

src/endpoints/url.py

@@ -8,7 +8,7 @@ from security import get_user_auth, get_token_or_user_auth
 from sqlalchemy.orm import Session
 
 from db import crud, schemas
-from db.database import get_db
+from db.database import get_db_dependency
 
 from worker import create_archive_task
 
@@ -32,23 +32,23 @@ def archive_url(archive: schemas.ArchiveCreate, email=Depends(get_token_or_user_
 def search_by_url(
         url: str, skip: int = 0, limit: int = 25,
         archived_after: datetime = None, archived_before: datetime = None,
-        db: Session = Depends(get_db),
+        db: Session = Depends(get_db_dependency),
         email=Depends(get_token_or_user_auth)):
     return crud.search_archives_by_url(db, url.strip(), email, skip=skip, limit=limit, archived_after=archived_after, archived_before=archived_before)
 
 
 @url_router.get("/latest", response_model=list[schemas.Archive], summary="Fetch latest URL archives for the authenticated user.")
-def latest(skip: int = 0, limit: int = 25, db: Session = Depends(get_db), email=Depends(get_user_auth)):
+def latest(skip: int = 0, limit: int = 25, db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)):
     return crud.search_archives_by_email(db, email, skip=skip, limit=limit)
 
 
 @url_router.get("/{id}", response_model=schemas.Archive, summary="Fetch a single URL archive by the associated id.")
-def lookup(id, db: Session = Depends(get_db), email=Depends(get_token_or_user_auth)):
+def lookup(id, db: Session = Depends(get_db_dependency), email=Depends(get_token_or_user_auth)):
     return crud.get_archive(db, id, email)
 
 
 @url_router.delete("/{id}", response_model=schemas.TaskDelete, summary="Delete a single URL archive by id.")
-def delete_task(id, db: Session = Depends(get_db), email=Depends(get_user_auth)):
+def delete_task(id, db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)):
     logger.info(f"deleting url archive task {id} request by {email}")
     #TODO: use response model?
     return JSONResponse({