delta-barth/delta-barth-py
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 3 Releases 3 Wiki Activity

prepare parsing for new pipeline format

Browse Source
This commit is contained in:
Florian Förster 2025-03-12 16:01:01 +01:00
parent 5bee01c5dd
commit cdcd613db0
1 changed files with 25 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/delta_barth/api/requests.py
View File

@ -8,19 +8,21 @@ from pydantic import BaseModel, PositiveInt, SkipValidation
from delta_barth.api.common import combine_route from delta_barth.api.common import combine_route
from delta_barth.errors import STATUS_HANDLER from delta_barth.errors import STATUS_HANDLER
from delta_barth.types import DelBarApiError, Status from delta_barth.types import DelBarApiError, ExportResponse, ResponseType, Status
if TYPE_CHECKING: if TYPE_CHECKING:
from delta_barth.api.common import Session from delta_barth.api.common import Session
from delta_barth.types import PipeResult
# ** sales data # ** sales data
# ** import
class SalesPrognosisRequestP(BaseModel): class SalesPrognosisRequestP(BaseModel):
FirmaId: SkipValidation[int | None] FirmaId: SkipValidation[int | None]
BuchungsDatum: SkipValidation[Datetime | None] BuchungsDatum: SkipValidation[Datetime | None]
class SalesPrognosisResponseEntry(BaseModel): class SalesPrognosisResponseEntry(ResponseType):
artikelId: PositiveInt artikelId: PositiveInt
firmaId: PositiveInt firmaId: PositiveInt
betrag: float # negative values are filtered out later betrag: float # negative values are filtered out later
@ -28,10 +30,27 @@ class SalesPrognosisResponseEntry(BaseModel):
buchungsDatum: Datetime buchungsDatum: Datetime
class SalesPrognosisResponse(BaseModel): class SalesPrognosisResponse(ResponseType):
daten: tuple[SalesPrognosisResponseEntry, ...] daten: tuple[SalesPrognosisResponseEntry, ...]
# ** export
class SalesPrognosisResult(BaseModel):
jahr: PositiveInt
monat: PositiveInt
betrag: float
vorhersage: float
class SalesPrognosisResults(BaseModel):
daten: tuple[SalesPrognosisResult, ...]
class SalesPrognosisResultsExport(ExportResponse):
response: SalesPrognosisResults
status: Status
def get_sales_prognosis_data( def get_sales_prognosis_data(
session: Session, session: Session,
company_id: int | None = None, company_id: int | None = None,
@ -62,3 +81,6 @@ def get_sales_prognosis_data(
status = STATUS_HANDLER.api_error(err) status = STATUS_HANDLER.api_error(err)
return response, status return response, status
def parse_sales_prognosis_response(pipe: PipeResult) -> ...: ...