64 lines
1.7 KiB
Python
64 lines
1.7 KiB
Python
from os import path
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
from fastapi.staticfiles import StaticFiles
|
|
from sqlmodel import Session, and_, select
|
|
from starlette import status
|
|
|
|
from .db import User, create_db_and_tables, engine, get_session
|
|
from .docker import client
|
|
|
|
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=status.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("/api/containers")
|
|
def get_containers(
|
|
session: Annotated[Session, Depends(get_session)],
|
|
credentials: Annotated[HTTPBasicCredentials, Depends(security)],
|
|
):
|
|
if credentials.username == "admin":
|
|
return [container.name for container in client.containers.list()]
|
|
|
|
return [
|
|
container.name
|
|
for container in client.containers.list(
|
|
filters={"label": f"owner={credentials.username}"}
|
|
)
|
|
]
|
|
|
|
|
|
app.mount(
|
|
"/",
|
|
StaticFiles(directory=f"{path.dirname(path.realpath(__file__))}/dist", html=True),
|
|
name="static",
|
|
)
|