[+] let payloads app reuse summarizer

1. add rest_client to transform app;
  2. wrap summarizer client inside payloads
    with domain and protocol configured via settings;
  3. update view to request summary generation per each
    input list inside POST:/payloads
This commit is contained in:
Siarhei Siniak 2025-07-25 13:07:06 +03:00
parent 17f5d81953
commit 38c9c838ed
4 changed files with 72 additions and 2 deletions

@ -4,6 +4,7 @@ import pydantic_settings
from typing import (
ClassVar,
Optional,
Literal,
)
@ -12,6 +13,9 @@ class Settings(pydantic_settings.BaseSettings):
_singleton: ClassVar[Optional['Settings']] = None
summarizer_domain: str
summarizer_protocol: Literal['http', 'https']
@classmethod
def singleton(cls) -> 'Settings':
if cls._singleton is None:

@ -0,0 +1,18 @@
from ..transform import rest_client as summarizer_rest_client
from ..transform.rest_client import SummaryRequest
from .settings import Settings as ModelSettings
from typing import (ClassVar, Optional,)
class SummarizerClient:
_summarizer_client : ClassVar[Optional[summarizer_rest_client.SummarizerClient]] = None
@classmethod
def singleton(cls) -> summarizer_rest_client.SummarizerClient:
if cls._summarizer_client is None:
cls._summarizer_client = summarizer_rest_client.SummarizerClient(
domain=ModelSettings.singleton().summarizer_domain,
protocol=ModelSettings.singleton().summarizer_protocol,
)
return cls._summarizer_client

@ -2,10 +2,11 @@ import fastapi
from typing import (Annotated, Any,)
from . import schema
from .summarizer import SummarizerClient, SummaryRequest
router = fastapi.APIRouter()
@router.post('payload')
@router.post('/payload')
async def payload_create(
list_1: Annotated[
list[str],
@ -16,9 +17,20 @@ async def payload_create(
fastapi.Body(),
],
) -> schema.Payload:
data_1 = await SummarizerClient.singleton().summarize_post(
request=SummaryRequest(
data=list_1
)
)
data_2 = await SummarizerClient.singleton().summarize_post(
request=SummaryRequest(
data=list_2
)
)
raise NotImplementedError
@router.get('payload/{paylaod_id}')
@router.get('/payload/{paylaod_id}')
async def payload_read(
payload_id: int,
) -> schema.Payload:

@ -0,0 +1,36 @@
import asyncio
import json
import aiohttp
import logging
from .schema import Summary, SummaryRequest
from typing import (Literal,)
logger = logging.getLogger(__name__)
class SummarizerClient:
def __init__(
self,
domain: str,
protocol: Literal['http', 'https'] = 'http',
) -> None:
self.domain = domain
self.protocol = protocol
async def summarize_post(
self,
request: SummaryRequest,
) -> Summary:
async with aiohttp.ClientSession() as session:
async with session.post(
'{}://{}/summarize'.format(self.protocol, self.domain),
json=json.loads(request.json()),
) as response:
data_json = await response.text()
data = Summary.model_validate_json(
data_json,
)
return data