mirror of
https://github.com/outbackdingo/step-ca-webui.git
synced 2026-01-27 10:20:25 +00:00
code style fix
This commit is contained in:
@@ -1,7 +1,16 @@
|
||||
import httpx
|
||||
from typing import List, Optional
|
||||
from datetime import date
|
||||
from shared.api_models import CertificateDTO, LogEntryDTO, CertificateGenerateRequest, CertificateGenerateResult, CertificateRenewResult, CertificateRevokeResult
|
||||
|
||||
import httpx
|
||||
|
||||
from shared.api_models import (
|
||||
CertificateDTO,
|
||||
LogEntryDTO,
|
||||
CertificateGenerateRequest,
|
||||
CertificateGenerateResult,
|
||||
CertificateRenewResult,
|
||||
CertificateRevokeResult,
|
||||
)
|
||||
|
||||
|
||||
class APIClient:
|
||||
def __init__(self, base_url: str):
|
||||
@@ -13,28 +22,46 @@ class APIClient:
|
||||
response.raise_for_status()
|
||||
return [CertificateDTO(**cert) for cert in response.json()]
|
||||
|
||||
async def generate_certificate(self, request: CertificateGenerateRequest) -> CertificateGenerateResult:
|
||||
response = await self.client.post("/certificates/generate", json=request.dict(), params={"preview": False})
|
||||
async def generate_certificate(
|
||||
self, request: CertificateGenerateRequest
|
||||
) -> CertificateGenerateResult:
|
||||
response = await self.client.post(
|
||||
"/certificates/generate", json=request.dict(), params={"preview": False}
|
||||
)
|
||||
response.raise_for_status()
|
||||
return CertificateGenerateResult(**response.json())
|
||||
|
||||
async def renew_certificate(self, cert_id: str, duration: int) -> CertificateRenewResult:
|
||||
response = await self.client.post("/certificates/renew", params={"certId": cert_id, "duration": duration, "preview": False})
|
||||
async def renew_certificate(
|
||||
self, cert_id: str, duration: int
|
||||
) -> CertificateRenewResult:
|
||||
response = await self.client.post(
|
||||
"/certificates/renew",
|
||||
params={"certId": cert_id, "duration": duration, "preview": False},
|
||||
)
|
||||
response.raise_for_status()
|
||||
return CertificateRenewResult(**response.json())
|
||||
|
||||
async def revoke_certificate(self, cert_id: str) -> CertificateRevokeResult:
|
||||
response = await self.client.post("/certificates/revoke", params={"certId": cert_id, "preview": False})
|
||||
response = await self.client.post(
|
||||
"/certificates/revoke", params={"certId": cert_id, "preview": False}
|
||||
)
|
||||
response.raise_for_status()
|
||||
return CertificateRevokeResult(**response.json())
|
||||
|
||||
async def get_logs(self, trace_id: Optional[str] = None, commands_only: bool = False, severity: List[str] = None, page: int = 1, page_size: int = 50) -> List[LogEntryDTO]:
|
||||
async def get_logs(
|
||||
self,
|
||||
trace_id: Optional[str] = None,
|
||||
commands_only: bool = False,
|
||||
severity: List[str] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> List[LogEntryDTO]:
|
||||
params = {
|
||||
"traceId": trace_id,
|
||||
"commandsOnly": commands_only,
|
||||
"severity": severity or ["DEBUG", "INFO", "WARN", "ERROR"],
|
||||
"page": page,
|
||||
"pageSize": page_size
|
||||
"pageSize": page_size,
|
||||
}
|
||||
response = await self.client.post("/logs", json=params)
|
||||
response.raise_for_status()
|
||||
|
||||
@@ -8,7 +8,6 @@ from fastapi.templating import Jinja2Templates
|
||||
from pydantic import BaseModel
|
||||
|
||||
from api_client import APIClient
|
||||
from shared.api_models import CertificateDTO, LogEntryDTO
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
@@ -18,6 +17,7 @@ templates = Jinja2Templates(directory="templates")
|
||||
|
||||
API_BASE_URL = "http://core-api:8000" # Adjust this to match your core API's URL
|
||||
|
||||
|
||||
async def get_api_client():
|
||||
client = APIClient(API_BASE_URL)
|
||||
try:
|
||||
@@ -25,6 +25,7 @@ async def get_api_client():
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
|
||||
class LogFilterData(BaseModel):
|
||||
commands_only: bool = False
|
||||
date_from: Optional[date] = None
|
||||
@@ -32,20 +33,26 @@ class LogFilterData(BaseModel):
|
||||
keywords: Optional[str] = None
|
||||
severity: List[str] = ["INFO", "WARN", "DEBUG", "ERROR"]
|
||||
|
||||
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
async def read_dashboard(request: Request, api_client: APIClient = Depends(get_api_client)):
|
||||
async def read_dashboard(
|
||||
request: Request, api_client: APIClient = Depends(get_api_client)
|
||||
):
|
||||
certificates = await api_client.list_certificates()
|
||||
return templates.TemplateResponse("dashboard.html.j2", {"request": request, "certificates": certificates})
|
||||
return templates.TemplateResponse(
|
||||
"dashboard.html.j2", {"request": request, "certificates": certificates}
|
||||
)
|
||||
|
||||
|
||||
@app.get("/logs", response_class=HTMLResponse)
|
||||
async def read_logs(
|
||||
request: Request,
|
||||
commands_only: bool = Query(False),
|
||||
date_from: Union[date, Literal[""], None] = Query(None),
|
||||
date_to: Union[date, Literal[""], None] = Query(None),
|
||||
keywords: Optional[str] = Query(None),
|
||||
severity: List[str] = Query(["INFO", "WARN", "DEBUG", "ERROR"]),
|
||||
api_client: APIClient = Depends(get_api_client)
|
||||
request: Request,
|
||||
commands_only: bool = Query(False),
|
||||
date_from: Union[date, Literal[""], None] = Query(None),
|
||||
date_to: Union[date, Literal[""], None] = Query(None),
|
||||
keywords: Optional[str] = Query(None),
|
||||
severity: List[str] = Query(["INFO", "WARN", "DEBUG", "ERROR"]),
|
||||
api_client: APIClient = Depends(get_api_client),
|
||||
):
|
||||
filter_data = LogFilterData(
|
||||
commands_only=commands_only,
|
||||
@@ -55,10 +62,12 @@ async def read_logs(
|
||||
severity=severity,
|
||||
)
|
||||
logs = await api_client.get_logs(
|
||||
commands_only=filter_data.commands_only,
|
||||
severity=filter_data.severity
|
||||
commands_only=filter_data.commands_only, severity=filter_data.severity
|
||||
)
|
||||
return templates.TemplateResponse("logs.html.j2", {"request": request, "logs": logs, "filter_data": filter_data})
|
||||
return templates.TemplateResponse(
|
||||
"logs.html.j2", {"request": request, "logs": logs, "filter_data": filter_data}
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
Reference in New Issue
Block a user