"""Secrets related endpoints factory.""" # pyright: reportUnusedFunction=false import logging from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, status from sshecret.backend.models import Secret from sshecret_admin.core.dependencies import AdminDependencies from sshecret_admin.services import AdminBackend from sshecret_admin.services.models import ( SecretCreate, SecretUpdate, SecretView, ) LOG = logging.getLogger(__name__) def create_router(dependencies: AdminDependencies) -> APIRouter: """Create secrets router.""" app = APIRouter(dependencies=[Depends(dependencies.get_current_active_user)]) @app.get("/secrets/") async def get_secret_names( admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)] ) -> list[Secret]: """Get Secret Names.""" return await admin.get_secrets() @app.post("/secrets/") async def add_secret( secret: SecretCreate, admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)], ) -> None: """Create a secret.""" await admin.add_secret(secret.name, secret.get_secret(), secret.clients) @app.get("/secrets/{name}") async def get_secret( name: str, admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)], ) -> SecretView: """Get a secret.""" secret_view = await admin.get_secret(name) if not secret_view: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Item not found." ) return secret_view @app.put("/secrets/{name}") async def update_secret( name: str, value: SecretUpdate, admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)], ) -> None: new_value = value.get_secret() await admin.update_secret(name, new_value) @app.delete("/secrets/{name}") async def delete_secret( name: str, admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)], ) -> None: """Delete secret.""" await admin.delete_secret(name) return app