delta-barth-py/src/delta_barth/api/common.py

from __future__ import annotations

import re
from datetime import datetime as Datetime
from typing import TYPE_CHECKING, Final

import requests
from pydantic import BaseModel, PositiveInt, SkipValidation
from requests import Response

from delta_barth._management import STATE_HANDLER
from delta_barth.constants import HTTP_CURRENT_CONNECTION
from delta_barth.errors import (
    ApiConnectionError,
    UnspecifiedRequestType,
)
from delta_barth.types import DelBarApiError, HttpRequestTypes

if TYPE_CHECKING:
    from delta_barth.types import Status

LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500))


def _assert_login_status() -> None:
    if not HTTP_CURRENT_CONNECTION.logged_in:
        raise ApiConnectionError("Curent session is not logged in")


def _strip_url_components(string: str) -> str:
    return re.sub(r"^[ /]+|[ /]+$", "", string)


def combine_route(base_url: str, route: str) -> str:
    base_url = _strip_url_components(base_url)
    route = _strip_url_components(route)
    return "/".join((base_url, route))


def ping(
    base_url: str,
    method: HttpRequestTypes,
) -> Response:
    ROUTE: Final[str] = "ping"
    URL: Final = combine_route(base_url, ROUTE)

    resp: Response
    if method == HttpRequestTypes.GET:
        resp = requests.get(URL)
    elif method == HttpRequestTypes.PUT:
        resp = requests.put(URL)
    elif method == HttpRequestTypes.DELETE:
        resp = requests.delete(URL)
    else:
        raise UnspecifiedRequestType(f"Request type {method} not defined for endpoint")

    return resp


# ** login
class LoginRequest(BaseModel):
    userName: str
    password: str
    databaseName: str
    mandantName: str


class LoginResponse(BaseModel):
    token: str


def login(
    base_url: str,
    user_name: str,
    password: str,
    database: str,
    mandant: str,
) -> tuple[LoginResponse, Status]:
    ROUTE: Final[str] = "user/login"
    URL: Final = combine_route(base_url, ROUTE)

    login_req = LoginRequest(
        userName=user_name,
        password=password,
        databaseName=database,
        mandantName=mandant,
    )
    resp = requests.put(
        URL,
        login_req.model_dump_json(),
        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore
    )

    response: LoginResponse
    status: Status
    if resp.status_code == 200:
        response = LoginResponse(**resp.json())
        status = STATE_HANDLER.pipe_states.SUCCESS
        HTTP_CURRENT_CONNECTION.add_session_token(response.token)
    else:
        response = LoginResponse(token="")
        err = DelBarApiError(status_code=resp.status_code, **resp.json())
        status = STATE_HANDLER.api_error(err)

    return response, status


# ** logout
def logout(
    base_url: str,
) -> tuple[None, Status]:
    ROUTE: Final[str] = "user/logout"
    URL: Final = combine_route(base_url, ROUTE)

    resp = requests.put(
        URL,
        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore
    )

    response = None
    status: Status
    if resp.status_code == 200:
        status = STATE_HANDLER.SUCCESS
        HTTP_CURRENT_CONNECTION.remove_session_token()
    else:
        err = DelBarApiError(status_code=resp.status_code, **resp.json())
        status = STATE_HANDLER.api_error(err)

    return response, status


# ** sales data
class SalesPrognosisRequestP(BaseModel):
    FirmaId: SkipValidation[int | None]
    BuchungsDatum: SkipValidation[Datetime | None]


class SalesPrognosisResponseEntry(BaseModel):
    artikelId: PositiveInt
    firmaId: PositiveInt
    betrag: float  # negative values are filtered out later
    menge: float  # reasons for negative values unknown
    buchungsDatum: Datetime


class SalesPrognosisResponse(BaseModel):
    daten: tuple[SalesPrognosisResponseEntry, ...]
    # error: DelBarApiError | None = None


def get_sales_prognosis_data(
    base_url: str,
    company_id: int | None = None,
    start_date: Datetime | None = None,
) -> tuple[SalesPrognosisResponse, Status]:
    _assert_login_status()
    ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
    URL: Final = combine_route(base_url, ROUTE)

    sales_prog_req = SalesPrognosisRequestP(
        FirmaId=company_id,
        BuchungsDatum=start_date,
    )
    resp = requests.get(
        URL,
        params=sales_prog_req.model_dump(mode="json", exclude_none=True),
        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore[argumentType]
    )

    response: SalesPrognosisResponse
    status: Status
    if resp.status_code == 200:
        response = SalesPrognosisResponse(**resp.json())
        status = STATE_HANDLER.SUCCESS
    else:
        response = SalesPrognosisResponse(daten=tuple())
        err = DelBarApiError(status_code=resp.status_code, **resp.json())
        status = STATE_HANDLER.api_error(err)

    # elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value:  # pragma: no cover
    #     err = DelBarApiError(status_code=resp.status_code, **resp.json())
    #     response = SalesPrognosisResponse(daten=tuple(), error=err)
    # else:  # pragma: no cover
    #     _raise_for_unknown_error(resp)

    return response, status