""" SuperFaktura API Client. This module provides classes and functions for working with the SuperFaktura API. It allows for reading, creating, updating, and deleting data in SuperFaktura. Classes: - SuperFakturaAPI: The base class for working with the SuperFaktura API. - SuperFakturaAPIException: An exception for errors when working with the SuperFaktura API. - SuperFakturaAPIMissingCredentialsException: An exception for missing login credentials. Functions: - get: Retrieves data from the SuperFaktura API. - post: Creates or updates data in the SuperFaktura API. Usage: import superfaktura.superfaktura_api # Create an instance of SuperFakturaAPI api = superfaktura.superfaktura_api.SuperFakturaAPI() # Retrieve data from the SuperFaktura API data = api.get('endpoint') # Create or update data in the SuperFaktura API api.post('endpoint', data) """ import os from typing import Dict import requests from dotenv import load_dotenv # type: ignore from superfaktura.enumerations.data_format import DataFormat class SuperFakturaAPIException(Exception): """Exception for errors when working with the SuperFaktura API.""" class SuperFakturaAPIMissingCredentialsException(Exception): """Exception for missing login credentials.""" class SuperFakturaAPI: """Base class for working with the SuperFaktura API.""" def __init__(self) -> None: load_dotenv() _api_key = os.getenv("SUPERFAKTURA_API_KEY") self._api_url = os.getenv("SUPERFAKTURA_API_URL") _api_mail = os.getenv("SUPERFAKTURA_API_EMAIL") _api_company_id = os.getenv("SUPERFAKTURA_API_COMPANY_ID") if not _api_key or not self._api_url or not _api_mail or not _api_company_id: raise SuperFakturaAPIMissingCredentialsException( "Please ensure, that necessary " "credentials are set. Please see" " README.md" ) self._auth_header = { "Authorization": f"SFAPI email={_api_mail}&apikey={_api_key}&company_id=" f"{_api_company_id}" } def get( self, endpoint: str, data_format: DataFormat = DataFormat.JSON, timeout: int = 5 ) -> Dict: """ Retrieves data from the SuperFaktura API. Retrieves data from the specified endpoint in the SuperFaktura API. Args: endpoint (str): The API endpoint to retrieve data from (e.g. 'invoices', 'clients', etc.). timeout (int, optional): The timeout for the API request in seconds. Defaults to 5. Returns: Dict: The retrieved data in JSON format. Raises: SuperFakturaAPIException: If the API request fails or returns an error. Examples: >>> api = SuperFakturaAPI() >>> data = api.get('invoices') >>> print(data) Notes: The available endpoints can be found in the SuperFaktura API documentation. """ url = f"{self._api_url}/{endpoint}" req = requests.get(url=url, headers=self._auth_header, timeout=timeout) if req.status_code == 200: if data_format == DataFormat.JSON: return req.json() elif data_format == DataFormat.PDF: return {"pdf": req.content} # returns a dict with the PDF content else: raise SuperFakturaAPIException("Invalid data format") raise SuperFakturaAPIException( f"Get status code: {req.status_code}; {req.json()}" ) def post(self, endpoint: str, data: str, timeout: int = 5) -> Dict: """ Creates or updates data in the SuperFaktura API. Creates or updates data in the specified endpoint in the SuperFaktura API. Args: endpoint (str): The API endpoint to create or update data in (e.g. 'invoices', 'clients', etc.). data (str): The data to be created or updated in JSON format. timeout (int, optional): The timeout for the API request in seconds. Defaults to 5. Returns: Dict: The created or updated data in JSON format. Raises: SuperFakturaAPIException: If the API request fails or returns an error. Examples: >>> api = SuperFakturaAPI() >>> data = '{"name": "Example Invoice", "amount": 100.0}' >>> response = api.post('invoices', data) >>> print(response) Notes: The available endpoints can be found in the SuperFaktura API documentation. The data should be a valid JSON string. """ url = f"{self._api_url}/{endpoint}" req = requests.post( url=url, headers=self._auth_header, data={"data": data}, timeout=timeout ) if req.status_code == 200: return req.json() raise SuperFakturaAPIException( f"Post status code: {req.status_code}; {req.json()}" )