From 8d5bf14be3a41007970ef6c61c67d579ef5d982f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Richard=20Kub=C3=AD=C4=8Dek?= Date: Tue, 4 Mar 2025 13:40:40 +0100 Subject: [PATCH] superfaktura_api: add download method for save IO[bytes] data stream --- examples/add_invoice.py | 7 ++--- superfaktura/invoice.py | 15 ++++++----- superfaktura/superfaktura_api.py | 44 +++++++++++++++++++++++++++++++- 3 files changed, 54 insertions(+), 12 deletions(-) diff --git a/examples/add_invoice.py b/examples/add_invoice.py index e085ad0..400f27f 100644 --- a/examples/add_invoice.py +++ b/examples/add_invoice.py @@ -21,8 +21,6 @@ Dependencies: - superfaktura.utils.data_types.Date """ -from pathlib import Path - from superfaktura.bank_account import BankAccount from superfaktura.client_contacts import ClientContactModel from superfaktura.enumerations.currency import Currencies @@ -71,10 +69,9 @@ def main(): ), invoice_settings=InvoiceSettings(language=Language.English), ) - _pdf = invoice.get_pdf(invoice=resp, language=Language.English) - p = Path("invoice.pdf") - p.write_bytes(_pdf) + with open("invoice.pdf", "wb") as f: + invoice.get_pdf(invoice=resp, descriptor=f, language=Language.English) if __name__ == "__main__": diff --git a/superfaktura/invoice.py b/superfaktura/invoice.py index da1caea..cbbd827 100644 --- a/superfaktura/invoice.py +++ b/superfaktura/invoice.py @@ -50,7 +50,7 @@ Usage: """ from dataclasses import dataclass, asdict -from typing import Optional, List +from typing import Optional, List, IO import json from superfaktura.client_contacts import ClientContactModel @@ -288,18 +288,21 @@ class Invoice(SuperFakturaAPI): return invoice_resp def get_pdf( - self, invoice: InvoiceRespModel, language: str = Language.Czech - ) -> bytes: + self, + invoice: InvoiceRespModel, + descriptor: IO[bytes], + language: str = Language.Czech, + ) -> None: """ Retrieves the PDF of the invoice. Args: invoice (InvoiceRespModel): The response model for the invoice. + descriptor (IO[bytes]): The descriptor to write the PDF data to. language (str): The language for the PDF. Returns: - bytes: A bytes containing the PDF data. + None """ url = f"{language}/invoices/pdf/{invoice.invoice_id}/token:{invoice.invoice_token}" - document = self.get(url) - return document + self.download(url, descriptor) diff --git a/superfaktura/superfaktura_api.py b/superfaktura/superfaktura_api.py index 65880d5..9e6e559 100644 --- a/superfaktura/superfaktura_api.py +++ b/superfaktura/superfaktura_api.py @@ -24,7 +24,7 @@ Usage: """ import os -from typing import Dict +from typing import Dict, IO import requests from dotenv import load_dotenv # type: ignore @@ -97,6 +97,48 @@ class SuperFakturaAPI: f"Get status code: {req.status_code}; {req.content!r}" ) + def download(self, endpoint: str, descriptor: IO[bytes], timeout: int = 5) -> None: + """ + Download data stream from the SuperFaktura API. + + Download data stream from the specified endpoint in the SuperFaktura API. + + Args: + endpoint (str): The API endpoint to retrieve data from (e.g. 'invoices', 'clients', + etc.). + descriptor (IO[bytes]): The descriptor to write the data stream to. + timeout (int, optional): The timeout for the API request in seconds. Defaults to 5. + + Returns: + None + + Raises: + SuperFakturaAPIException: If the API request fails or returns an error. + + Examples: + >>> from superfaktura.invoice import Invoice + >>> from superfaktura.enumerations.language import Language + >>> invoice = Invoice() + >>> with open("invoice.pdf", "wb") as f: + >>> invoice.get_pdf(invoice=invoice_data, descriptor=f, language=Language.English) + + 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 descriptor.writable(): + descriptor.write(req.content) + else: + raise SuperFakturaAPIException( + f"Descriptor {descriptor} is not writable" + ) + else: + raise SuperFakturaAPIException( + f"Download status code: {req.status_code}; {req.content!r}" + ) + def post(self, endpoint: str, data: str, timeout: int = 5) -> Dict: """ Creates or updates data in the SuperFaktura API.