mirror of
https://github.com/msoedov/agentic_security.git
synced 2026-06-29 08:39:57 +02:00
Add type annotations to functions and methods for improved clarity and maintainabiliy
This commit is contained in:
@@ -4,11 +4,18 @@ from asyncio import Event, Queue
|
||||
from fastapi import FastAPI
|
||||
from fastapi.responses import ORJSONResponse
|
||||
|
||||
from agentic_security.http_spec import LLMSpec
|
||||
from typing import Any, Dict
|
||||
|
||||
tools_inbox: Queue = Queue()
|
||||
stop_event: Event = Event()
|
||||
current_run: str = {"spec": "", "id": ""}
|
||||
_secrets = {}
|
||||
_secrets: dict[str, str] = {}
|
||||
|
||||
current_run: Dict[str, int | LLMSpec] = {
|
||||
"spec": "",
|
||||
"id": ""
|
||||
}
|
||||
|
||||
def create_app() -> FastAPI:
|
||||
"""Create and configure the FastAPI application."""
|
||||
@@ -26,29 +33,29 @@ def get_stop_event() -> Event:
|
||||
return stop_event
|
||||
|
||||
|
||||
def get_current_run() -> str:
|
||||
def get_current_run() -> Dict[str, int | LLMSpec]:
|
||||
"""Get the current run id."""
|
||||
return current_run
|
||||
|
||||
|
||||
def set_current_run(spec):
|
||||
def set_current_run(spec : LLMSpec) -> Dict[str, int | LLMSpec]:
|
||||
"""Set the current run id."""
|
||||
current_run["id"] = hash(id(spec))
|
||||
current_run["spec"] = spec
|
||||
return current_run
|
||||
|
||||
|
||||
def get_secrets():
|
||||
def get_secrets() -> dict[str, str]:
|
||||
return _secrets
|
||||
|
||||
|
||||
def set_secrets(secrets):
|
||||
def set_secrets(secrets : dict[str, str]) -> dict[str, str]:
|
||||
_secrets.update(secrets)
|
||||
expand_secrets(_secrets)
|
||||
return _secrets
|
||||
|
||||
|
||||
def expand_secrets(secrets):
|
||||
def expand_secrets(secrets : dict[str, str]) -> None:
|
||||
for key in secrets:
|
||||
val = secrets[key]
|
||||
if val.startswith("$"):
|
||||
|
||||
Reference in New Issue
Block a user