39 lines
1.0 KiB
Python
39 lines
1.0 KiB
Python
"""Shared application state for the data-ingestion service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional
|
|
|
|
import nats
|
|
import redis.asyncio as redis
|
|
|
|
from config import Settings
|
|
from rapidapi_client import RapidAPIClient
|
|
from apillama_processor import APILLAMAProcessor
|
|
from openapi_parser import OpenAPIParser
|
|
from tool_generator import ToolGenerator
|
|
|
|
|
|
@dataclass
|
|
class ServiceState:
|
|
"""Container for runtime dependencies."""
|
|
|
|
settings: Settings = field(default_factory=Settings)
|
|
redis_client: Optional[redis.Redis] = None
|
|
nats_client: Optional[nats.NATS] = None
|
|
rapidapi_client: Optional[RapidAPIClient] = None
|
|
apillama_processor: Optional[APILLAMAProcessor] = None
|
|
openapi_parser: Optional[OpenAPIParser] = None
|
|
tool_generator: Optional[ToolGenerator] = None
|
|
background_tasks: List[asyncio.Task] = field(default_factory=list)
|
|
|
|
|
|
_state = ServiceState()
|
|
|
|
|
|
def get_state() -> ServiceState:
|
|
"""Return the singleton service state."""
|
|
return _state
|