mdrs-client-python/mdrsclient/api/base.py

from abc import ABC

import requests
from pydantic import parse_obj_as
from requests import Response

from mdrsclient.connection import MDRSConnection
from mdrsclient.exceptions import (
    BadRequestException,
    ForbiddenException,
    UnauthorizedException,
    UnexpectedException,
)
from mdrsclient.models import DRFStandardizedErrors


class BaseApi(ABC):
    connection: MDRSConnection

    def __init__(self, connection: MDRSConnection) -> None:
        self.connection = connection

    def _get(self, url, *args, **kwargs) -> Response:
        return self.connection.session.get(self.__build_url(url), *args, **kwargs)

    def _post(self, url, *args, **kwargs) -> Response:
        return self.connection.session.post(self.__build_url(url), *args, **kwargs)

    def _put(self, url, *args, **kwargs) -> Response:
        return self.connection.session.put(self.__build_url(url), *args, **kwargs)

    def _delete(self, url, *args, **kwargs) -> Response:
        return self.connection.session.delete(self.__build_url(url), *args, **kwargs)

    def _patch(self, url, *args, **kwargs) -> Response:
        return self.connection.session.patch(self.__build_url(url), *args, **kwargs)

    def _raise_response_error(self, response: Response) -> None:
        if response.status_code >= 300:
            if response.status_code < 400 or response.status_code >= 500:
                raise UnexpectedException(f"Unexpected status code returned: {response.status_code}.")
            errors = parse_obj_as(DRFStandardizedErrors, response.json())
            if response.status_code == requests.codes.bad_request:
                raise BadRequestException(errors.errors[0].detail)
            elif response.status_code == requests.codes.unauthorized:
                raise UnauthorizedException("Login required.")
            elif response.status_code == requests.codes.forbidden:
                raise ForbiddenException("You do not have enough permissions. Access is denied.")
            else:
                raise UnexpectedException(errors.errors[0].detail)

    def __build_url(self, *args: tuple) -> str:
        return self.connection.build_url(*args)