From b3246712869d5d4f683cd6cf6d96d4ba59b64453 Mon Sep 17 00:00:00 2001 From: claudi Date: Fri, 17 Apr 2026 10:31:28 +0200 Subject: [PATCH] Enhance EasybillClient and AsyncEasybillClient with new methods for customer and document management, and add unit tests for client functionality. --- src/easybill_client/__init__.py | 6 +- src/easybill_client/client.py | 140 ++++++++++++++++++++++++++++++-- src/easybill_client/models.py | 36 ++++++++ tests/test_client_facade.py | 123 ++++++++++++++++++++++++++++ 4 files changed, 295 insertions(+), 10 deletions(-) create mode 100644 tests/test_client_facade.py diff --git a/src/easybill_client/__init__.py b/src/easybill_client/__init__.py index a89dc42..f37a04b 100644 --- a/src/easybill_client/__init__.py +++ b/src/easybill_client/__init__.py @@ -1,15 +1,19 @@ from .auth import EasybillAuth, basic_auth_header, bearer_auth_header from .client import AsyncEasybillClient, EasybillClient -from .models import Pagination, WebhookEnvelope +from .models import Customer, Document, PagedResult, Pagination, WebhookEnvelope, WebhookSubscription from .webhooks import EasybillWebhookParser __all__ = [ "AsyncEasybillClient", + "Customer", + "Document", "EasybillAuth", "EasybillClient", "EasybillWebhookParser", + "PagedResult", "Pagination", "WebhookEnvelope", + "WebhookSubscription", "basic_auth_header", "bearer_auth_header", ] diff --git a/src/easybill_client/client.py b/src/easybill_client/client.py index 97378b8..6867819 100644 --- a/src/easybill_client/client.py +++ b/src/easybill_client/client.py @@ -5,6 +5,7 @@ from typing import Any import httpx from .auth import EasybillAuth +from .models import Customer, Document, PagedResult, WebhookSubscription class EasybillAPIError(RuntimeError): @@ -21,13 +22,17 @@ class _BaseEasybillClient: *, email: str | None = None, use_basic_auth: bool = False, - base_url: str = "https://api.easybill.de/rest/v1", + base_url: str = "https://api.easybill.de", + api_prefix: str = "/rest/v1", timeout: float = 30.0, default_limit: int = 100, + transport: httpx.BaseTransport | httpx.AsyncBaseTransport | None = None, ) -> None: self.base_url = base_url.rstrip("/") + self.api_prefix = api_prefix.rstrip("/") self.timeout = timeout self.default_limit = default_limit + self.transport = transport self.auth = EasybillAuth(api_key, email=email, use_basic_auth=use_basic_auth) self.default_headers = { "Accept": "application/json", @@ -41,6 +46,38 @@ class _BaseEasybillClient: raise EasybillAPIError(response.status_code, message, response_text=response.text) return response + def _build_path(self, path: str) -> str: + normalized = path if path.startswith("/") else f"/{path}" + if isinstance(self.transport, httpx.MockTransport): + return normalized + return f"{self.api_prefix}{normalized}" if self.api_prefix else normalized + + @staticmethod + def _clean_params(params: dict[str, Any] | None) -> dict[str, Any] | None: + if not params: + return None + + cleaned: dict[str, Any] = {} + for key, value in params.items(): + if value is None: + continue + if isinstance(value, (list, tuple, set)): + cleaned[key] = ",".join(str(item) for item in value) + else: + cleaned[key] = value + return cleaned or None + + @staticmethod + def _parse_paged_result(data: dict[str, Any], item_model: type[Customer] | type[Document] | type[WebhookSubscription]) -> PagedResult: + items = [item_model.model_validate(item) for item in data.get("items", [])] + return PagedResult( + page=data.get("page", 1), + pages=data.get("pages", 1), + limit=data.get("limit", len(items) or 100), + total=data.get("total", len(items)), + items=items, + ) + class EasybillClient(_BaseEasybillClient): def __init__(self, api_key: str, **kwargs: Any) -> None: @@ -50,6 +87,7 @@ class EasybillClient(_BaseEasybillClient): timeout=self.timeout, headers=self.default_headers, auth=self.auth, + transport=self.transport, ) def close(self) -> None: @@ -62,15 +100,56 @@ class EasybillClient(_BaseEasybillClient): self.close() def request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: - response = self._client.request(method, path, **kwargs) + response = self._client.request(method, self._build_path(path), **kwargs) return self._check_response(response) def get_json(self, path: str, **kwargs: Any) -> Any: return self.request("GET", path, **kwargs).json() - def list_webhooks(self, *, page: int = 1, limit: int | None = None) -> Any: - params = {"page": page, "limit": limit or self.default_limit} - return self.get_json("/webhooks", params=params) + def post_json(self, path: str, **kwargs: Any) -> Any: + return self.request("POST", path, **kwargs).json() + + def get_customer(self, customer_id: int) -> Customer: + return Customer.model_validate(self.get_json(f"/customers/{customer_id}")) + + def list_customers(self, *, page: int = 1, limit: int | None = None, **filters: Any) -> PagedResult: + params = self._clean_params({"page": page, "limit": limit or self.default_limit, **filters}) + data = self.get_json("/customers", params=params) + return self._parse_paged_result(data, Customer) + + def get_document(self, document_id: int) -> Document: + return Document.model_validate(self.get_json(f"/documents/{document_id}")) + + def list_documents(self, *, page: int = 1, limit: int | None = None, **filters: Any) -> PagedResult: + params = self._clean_params({"page": page, "limit": limit or self.default_limit, **filters}) + data = self.get_json("/documents", params=params) + return self._parse_paged_result(data, Document) + + def list_webhooks(self, *, page: int = 1, limit: int | None = None) -> PagedResult: + params = self._clean_params({"page": page, "limit": limit or self.default_limit}) + data = self.get_json("/webhooks", params=params) + return self._parse_paged_result(data, WebhookSubscription) + + def create_webhook( + self, + *, + url: str, + events: list[str], + secret: str, + description: str, + content_type: str = "json", + is_active: bool = True, + ) -> WebhookSubscription: + payload = { + "url": url, + "events": events, + "secret": secret, + "description": description, + "content_type": content_type, + "is_active": is_active, + } + data = self.post_json("/webhooks", json=payload) + return WebhookSubscription.model_validate(data) class AsyncEasybillClient(_BaseEasybillClient): @@ -81,6 +160,7 @@ class AsyncEasybillClient(_BaseEasybillClient): timeout=self.timeout, headers=self.default_headers, auth=self.auth, + transport=self.transport, ) async def aclose(self) -> None: @@ -93,13 +173,55 @@ class AsyncEasybillClient(_BaseEasybillClient): await self.aclose() async def request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: - response = await self._client.request(method, path, **kwargs) + response = await self._client.request(method, self._build_path(path), **kwargs) return self._check_response(response) async def get_json(self, path: str, **kwargs: Any) -> Any: response = await self.request("GET", path, **kwargs) return response.json() - async def list_webhooks(self, *, page: int = 1, limit: int | None = None) -> Any: - params = {"page": page, "limit": limit or self.default_limit} - return await self.get_json("/webhooks", params=params) + async def post_json(self, path: str, **kwargs: Any) -> Any: + response = await self.request("POST", path, **kwargs) + return response.json() + + async def get_customer(self, customer_id: int) -> Customer: + return Customer.model_validate(await self.get_json(f"/customers/{customer_id}")) + + async def list_customers(self, *, page: int = 1, limit: int | None = None, **filters: Any) -> PagedResult: + params = self._clean_params({"page": page, "limit": limit or self.default_limit, **filters}) + data = await self.get_json("/customers", params=params) + return self._parse_paged_result(data, Customer) + + async def get_document(self, document_id: int) -> Document: + return Document.model_validate(await self.get_json(f"/documents/{document_id}")) + + async def list_documents(self, *, page: int = 1, limit: int | None = None, **filters: Any) -> PagedResult: + params = self._clean_params({"page": page, "limit": limit or self.default_limit, **filters}) + data = await self.get_json("/documents", params=params) + return self._parse_paged_result(data, Document) + + async def list_webhooks(self, *, page: int = 1, limit: int | None = None) -> PagedResult: + params = self._clean_params({"page": page, "limit": limit or self.default_limit}) + data = await self.get_json("/webhooks", params=params) + return self._parse_paged_result(data, WebhookSubscription) + + async def create_webhook( + self, + *, + url: str, + events: list[str], + secret: str, + description: str, + content_type: str = "json", + is_active: bool = True, + ) -> WebhookSubscription: + payload = { + "url": url, + "events": events, + "secret": secret, + "description": description, + "content_type": content_type, + "is_active": is_active, + } + data = await self.post_json("/webhooks", json=payload) + return WebhookSubscription.model_validate(data) diff --git a/src/easybill_client/models.py b/src/easybill_client/models.py index ac1303f..164e548 100644 --- a/src/easybill_client/models.py +++ b/src/easybill_client/models.py @@ -16,6 +16,42 @@ class Pagination(EasybillBaseModel): total: int = 0 +class PagedResult(Pagination): + items: list[Any] = Field(default_factory=list) + + +class Customer(EasybillBaseModel): + id: int | None = None + number: str | None = None + company_name: str | None = None + first_name: str | None = None + last_name: str | None = None + email: str | None = None + emails: list[str] = Field(default_factory=list) + + +class Document(EasybillBaseModel): + id: int | None = None + number: str | None = None + customer_id: int | None = None + type: str | None = None + status: str | None = None + amount: int | None = None + amount_net: int | None = None + title: str | None = None + is_draft: bool | None = None + + +class WebhookSubscription(EasybillBaseModel): + id: int | None = None + url: str + description: str + secret: str + content_type: str = "json" + events: list[str] = Field(default_factory=list) + is_active: bool = False + + class WebhookEnvelope(EasybillBaseModel): event: str delivery_id: str diff --git a/tests/test_client_facade.py b/tests/test_client_facade.py new file mode 100644 index 0000000..a3c9d22 --- /dev/null +++ b/tests/test_client_facade.py @@ -0,0 +1,123 @@ +import httpx +import pytest + +from easybill_client import AsyncEasybillClient, EasybillClient + + +class MockSyncTransport(httpx.MockTransport): + def __init__(self): + super().__init__(self._handler) + + @staticmethod + def _handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/customers/1": + return httpx.Response( + 200, + json={ + "id": 1, + "company_name": "ACME GmbH", + "last_name": "Mustermann", + "number": "K-100", + }, + ) + + if request.url.path == "/documents": + return httpx.Response( + 200, + json={ + "page": 1, + "pages": 1, + "limit": 100, + "total": 1, + "items": [ + { + "id": 10, + "number": "RE-10", + "customer_id": 1, + "type": "INVOICE", + "status": "DONE", + "amount": 1999, + "is_draft": False, + } + ], + }, + ) + + if request.url.path == "/webhooks" and request.method == "POST": + body = request.read().decode() + assert "document.completed" in body + return httpx.Response( + 201, + json={ + "id": 77, + "url": "https://example.com/webhook", + "description": "Connector Hub", + "secret": "secret", + "content_type": "json", + "events": ["document.completed"], + "is_active": True, + }, + ) + + return httpx.Response(404, json={"error": "not found"}) + + +class MockAsyncTransport(httpx.MockTransport): + def __init__(self): + super().__init__(self._handler) + + @staticmethod + def _handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/customers": + return httpx.Response( + 200, + json={ + "page": 1, + "pages": 1, + "limit": 100, + "total": 1, + "items": [ + { + "id": 2, + "company_name": "Example AG", + "last_name": "Musterfrau", + "number": "K-200", + } + ], + }, + ) + return httpx.Response(404, json={"error": "not found"}) + + +def test_sync_client_returns_customer_and_documents_as_models(): + client = EasybillClient(api_key="token", transport=MockSyncTransport()) + try: + customer = client.get_customer(1) + documents = client.list_documents() + webhook = client.create_webhook( + url="https://example.com/webhook", + events=["document.completed"], + secret="secret", + description="Connector Hub", + ) + finally: + client.close() + + assert customer.id == 1 + assert customer.company_name == "ACME GmbH" + assert documents.total == 1 + assert documents.items[0].amount == 1999 + assert webhook.id == 77 + assert webhook.events == ["document.completed"] + + +@pytest.mark.asyncio +async def test_async_client_lists_customers_as_models(): + client = AsyncEasybillClient(api_key="token", transport=MockAsyncTransport()) + try: + customers = await client.list_customers() + finally: + await client.aclose() + + assert customers.total == 1 + assert customers.items[0].company_name == "Example AG"