From d17babab0b5167e5e231fa86238d3257a6e2c2ab Mon Sep 17 00:00:00 2001 From: stone-24tch3r <100294019+stone-w4tch3r@users.noreply.github.com> Date: Fri, 27 Sep 2024 23:31:32 +0500 Subject: [PATCH] code style fix --- front/api_client.py | 47 +++++++++++++++++++++++++++++++++++---------- front/main.py | 35 ++++++++++++++++++++------------- 2 files changed, 59 insertions(+), 23 deletions(-) diff --git a/front/api_client.py b/front/api_client.py index b419e63..12b49ba 100644 --- a/front/api_client.py +++ b/front/api_client.py @@ -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() diff --git a/front/main.py b/front/main.py index f485fb7..4ed92db 100644 --- a/front/main.py +++ b/front/main.py @@ -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