43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
|
from typing import Annotated
|
||
|
|
|
||
|
|
from fastapi import Depends, FastAPI, HTTPException
|
||
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||
|
|
from sqlmodel import Session, and_, select
|
||
|
|
from starlette.status import HTTP_403_FORBIDDEN
|
||
|
|
|
||
|
|
from .db import User, create_db_and_tables, engine, get_session
|
||
|
|
|
||
|
|
security = HTTPBasic()
|
||
|
|
|
||
|
|
|
||
|
|
def check_auth(credentials: Annotated[HTTPBasicCredentials, Depends(security)]) -> User:
|
||
|
|
with Session(engine) as session:
|
||
|
|
statement = select(User).where(
|
||
|
|
and_(
|
||
|
|
User.username == credentials.username,
|
||
|
|
User.password == credentials.password,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
user = session.exec(statement).first()
|
||
|
|
|
||
|
|
if not user:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=HTTP_403_FORBIDDEN,
|
||
|
|
detail="Invalid authentication credentials",
|
||
|
|
)
|
||
|
|
|
||
|
|
return user
|
||
|
|
|
||
|
|
|
||
|
|
app = FastAPI(dependencies=[Depends(get_session), Depends(check_auth)])
|
||
|
|
|
||
|
|
|
||
|
|
@app.on_event("startup")
|
||
|
|
def on_startup() -> None:
|
||
|
|
create_db_and_tables()
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/")
|
||
|
|
def hello_world() -> str:
|
||
|
|
return "<p>Hello, World!</p>"
|