superfaktura_api: add download method for save IO[bytes] data stream

This commit is contained in:
Richard Kubíček
2025-03-04 13:40:40 +01:00
parent bb6b6343b0
commit 8d5bf14be3
3 changed files with 54 additions and 12 deletions

View File

@@ -21,8 +21,6 @@ Dependencies:
- superfaktura.utils.data_types.Date - superfaktura.utils.data_types.Date
""" """
from pathlib import Path
from superfaktura.bank_account import BankAccount from superfaktura.bank_account import BankAccount
from superfaktura.client_contacts import ClientContactModel from superfaktura.client_contacts import ClientContactModel
from superfaktura.enumerations.currency import Currencies from superfaktura.enumerations.currency import Currencies
@@ -71,10 +69,9 @@ def main():
), ),
invoice_settings=InvoiceSettings(language=Language.English), invoice_settings=InvoiceSettings(language=Language.English),
) )
_pdf = invoice.get_pdf(invoice=resp, language=Language.English)
p = Path("invoice.pdf") with open("invoice.pdf", "wb") as f:
p.write_bytes(_pdf) invoice.get_pdf(invoice=resp, descriptor=f, language=Language.English)
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -50,7 +50,7 @@ Usage:
""" """
from dataclasses import dataclass, asdict from dataclasses import dataclass, asdict
from typing import Optional, List from typing import Optional, List, IO
import json import json
from superfaktura.client_contacts import ClientContactModel from superfaktura.client_contacts import ClientContactModel
@@ -288,18 +288,21 @@ class Invoice(SuperFakturaAPI):
return invoice_resp return invoice_resp
def get_pdf( def get_pdf(
self, invoice: InvoiceRespModel, language: str = Language.Czech self,
) -> bytes: invoice: InvoiceRespModel,
descriptor: IO[bytes],
language: str = Language.Czech,
) -> None:
""" """
Retrieves the PDF of the invoice. Retrieves the PDF of the invoice.
Args: Args:
invoice (InvoiceRespModel): The response model for the invoice. 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. language (str): The language for the PDF.
Returns: Returns:
bytes: A bytes containing the PDF data. None
""" """
url = f"{language}/invoices/pdf/{invoice.invoice_id}/token:{invoice.invoice_token}" url = f"{language}/invoices/pdf/{invoice.invoice_id}/token:{invoice.invoice_token}"
document = self.get(url) self.download(url, descriptor)
return document

View File

@@ -24,7 +24,7 @@ Usage:
""" """
import os import os
from typing import Dict from typing import Dict, IO
import requests import requests
from dotenv import load_dotenv # type: ignore from dotenv import load_dotenv # type: ignore
@@ -97,6 +97,48 @@ class SuperFakturaAPI:
f"Get status code: {req.status_code}; {req.content!r}" 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: 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 SuperFaktura API.