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

from __future__ import annotations

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

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

from delta_barth.constants import HTTP_CURRENT_CONNECTION, KnownDelBarApiErrorCodes
from delta_barth.errors import (
    ApiConnectionError,
    UnknownApiErrorCode,
    UnspecifiedRequestType,
)
from delta_barth.types import HttpRequestTypes

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


class DelBarApiError(BaseModel):
    status_code: int
    message: str = ""
    code: str | None = None
    hints: str | None = None
    type: str | None = None
    title: str | None = None
    errors: dict | None = None
    traceId: str | None = None


def _raise_for_unknown_error(
    resp: Response,
) -> Never:
    raise UnknownApiErrorCode(
        f"Unknown response code for request. Status Code: {resp.status_code}, "
        f"Content: {resp.text}"
    )


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
    error: DelBarApiError | None = None


def login(
    base_url: str,
    user_name: str,
    password: str,
    database: str,
    mandant: str,
) -> LoginResponse:
    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
    if resp.status_code == 200:
        response = LoginResponse(**resp.json())
        HTTP_CURRENT_CONNECTION.add_session_token(response.token)
    elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value:
        err = DelBarApiError(status_code=resp.status_code, **resp.json())
        response = LoginResponse(token="", error=err)
    else:  # pragma: no cover
        _raise_for_unknown_error(resp)

    return response


# ** logout
class LogoutResponse(BaseModel):
    error: DelBarApiError | None = None


def logout(
    base_url: str,
) -> LogoutResponse:
    ROUTE: Final[str] = "user/logout"
    URL: Final = combine_route(base_url, ROUTE)

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

    response: LogoutResponse
    if resp.status_code == 200:
        response = LogoutResponse()
        HTTP_CURRENT_CONNECTION.remove_session_token()
    elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value:
        err = DelBarApiError(status_code=resp.status_code, **resp.json())
        response = LogoutResponse(error=err)
    else:  # pragma: no cover
        _raise_for_unknown_error(resp)

    return response


# ** 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,
) -> SalesPrognosisResponse:
    _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
    if resp.status_code == 200:
        response = SalesPrognosisResponse(**resp.json())
    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