diff --git a/LICENSE b/LICENSE index c4de03f..3da46e8 100644 --- a/LICENSE +++ b/LICENSE @@ -19,3 +19,4 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + diff --git a/pyproject.toml b/pyproject.toml index 9f8827e..6d8f949 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dev = [ "responses>=0.23.0", "ruff>=0.1.0", "mypy>=1.0.0", + "types-requests>=2.28.0", ] [project.urls] diff --git a/src/pobo/__init__.py b/src/pobo/__init__.py index 5d4dfc7..7e1d01f 100644 --- a/src/pobo/__init__.py +++ b/src/pobo/__init__.py @@ -5,21 +5,21 @@ """ from pobo.client import PoboClient -from pobo.webhook_handler import WebhookHandler -from pobo.enums import Language, WebhookEvent -from pobo.exceptions import ApiError, ValidationError, WebhookError, PoboError from pobo.dto import ( - LocalizedString, - Content, - Product, - Category, Blog, - Parameter, - ParameterValue, + Category, + Content, ImportResult, + LocalizedString, PaginatedResponse, + Parameter, + ParameterValue, + Product, WebhookPayload, ) +from pobo.enums import Language, WebhookEvent +from pobo.exceptions import ApiError, PoboError, ValidationError, WebhookError +from pobo.webhook_handler import WebhookHandler __version__ = "1.0.4" diff --git a/src/pobo/client.py b/src/pobo/client.py index 4fcfdd3..5a4860a 100644 --- a/src/pobo/client.py +++ b/src/pobo/client.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, Iterator, Type, TypeVar +from typing import Any, Dict, Iterator, List, Optional, Type, TypeVar, Union import requests @@ -43,28 +43,28 @@ def __init__( # Import methods - def import_products(self, products: list[Product | dict[str, Any]]) -> ImportResult: + def import_products(self, products: List[Union[Product, Dict[str, Any]]]) -> ImportResult: """Bulk import products. Maximum 100 items per request.""" self._validate_bulk_size(products) payload = [p.to_api_dict() if isinstance(p, Product) else p for p in products] response = self._request("POST", "/api/v2/rest/products", payload) return ImportResult.model_validate(response) - def import_categories(self, categories: list[Category | dict[str, Any]]) -> ImportResult: + def import_categories(self, categories: List[Union[Category, Dict[str, Any]]]) -> ImportResult: """Bulk import categories. Maximum 100 items per request.""" self._validate_bulk_size(categories) payload = [c.to_api_dict() if isinstance(c, Category) else c for c in categories] response = self._request("POST", "/api/v2/rest/categories", payload) return ImportResult.model_validate(response) - def import_parameters(self, parameters: list[Parameter | dict[str, Any]]) -> ImportResult: + def import_parameters(self, parameters: List[Union[Parameter, Dict[str, Any]]]) -> ImportResult: """Bulk import parameters. Maximum 100 items per request.""" self._validate_bulk_size(parameters) payload = [p.to_api_dict() if isinstance(p, Parameter) else p for p in parameters] response = self._request("POST", "/api/v2/rest/parameters", payload) return ImportResult.model_validate(response) - def import_blogs(self, blogs: list[Blog | dict[str, Any]]) -> ImportResult: + def import_blogs(self, blogs: List[Union[Blog, Dict[str, Any]]]) -> ImportResult: """Bulk import blogs. Maximum 100 items per request.""" self._validate_bulk_size(blogs) payload = [b.to_api_dict() if isinstance(b, Blog) else b for b in blogs] @@ -75,10 +75,10 @@ def import_blogs(self, blogs: list[Blog | dict[str, Any]]) -> ImportResult: def get_products( self, - page: int | None = None, - per_page: int | None = None, - last_update_from: datetime | None = None, - is_edited: bool | None = None, + page: Optional[int] = None, + per_page: Optional[int] = None, + last_update_from: Optional[datetime] = None, + is_edited: Optional[bool] = None, ) -> PaginatedResponse[Product]: """Get paginated list of products.""" params = self._build_query_params(page, per_page, last_update_from, is_edited) @@ -87,10 +87,10 @@ def get_products( def get_categories( self, - page: int | None = None, - per_page: int | None = None, - last_update_from: datetime | None = None, - is_edited: bool | None = None, + page: Optional[int] = None, + per_page: Optional[int] = None, + last_update_from: Optional[datetime] = None, + is_edited: Optional[bool] = None, ) -> PaginatedResponse[Category]: """Get paginated list of categories.""" params = self._build_query_params(page, per_page, last_update_from, is_edited) @@ -99,10 +99,10 @@ def get_categories( def get_blogs( self, - page: int | None = None, - per_page: int | None = None, - last_update_from: datetime | None = None, - is_edited: bool | None = None, + page: Optional[int] = None, + per_page: Optional[int] = None, + last_update_from: Optional[datetime] = None, + is_edited: Optional[bool] = None, ) -> PaginatedResponse[Blog]: """Get paginated list of blogs.""" params = self._build_query_params(page, per_page, last_update_from, is_edited) @@ -113,24 +113,24 @@ def get_blogs( def iter_products( self, - last_update_from: datetime | None = None, - is_edited: bool | None = None, + last_update_from: Optional[datetime] = None, + is_edited: Optional[bool] = None, ) -> Iterator[Product]: """Iterate through all products, handling pagination automatically.""" yield from self._iterate(self.get_products, last_update_from, is_edited) def iter_categories( self, - last_update_from: datetime | None = None, - is_edited: bool | None = None, + last_update_from: Optional[datetime] = None, + is_edited: Optional[bool] = None, ) -> Iterator[Category]: """Iterate through all categories, handling pagination automatically.""" yield from self._iterate(self.get_categories, last_update_from, is_edited) def iter_blogs( self, - last_update_from: datetime | None = None, - is_edited: bool | None = None, + last_update_from: Optional[datetime] = None, + is_edited: Optional[bool] = None, ) -> Iterator[Blog]: """Iterate through all blogs, handling pagination automatically.""" yield from self._iterate(self.get_blogs, last_update_from, is_edited) @@ -140,8 +140,8 @@ def iter_blogs( def _iterate( self, get_method: Any, - last_update_from: datetime | None, - is_edited: bool | None, + last_update_from: Optional[datetime], + is_edited: Optional[bool], ) -> Iterator[Any]: """Generic iterator for paginated responses.""" page = 1 @@ -157,7 +157,7 @@ def _iterate( break page += 1 - def _validate_bulk_size(self, items: list[Any]) -> None: + def _validate_bulk_size(self, items: List[Any]) -> None: """Validate bulk import size.""" if not items: raise ValidationError.empty_payload() @@ -166,13 +166,13 @@ def _validate_bulk_size(self, items: list[Any]) -> None: def _build_query_params( self, - page: int | None, - per_page: int | None, - last_update_from: datetime | None, - is_edited: bool | None, - ) -> dict[str, Any]: + page: Optional[int], + per_page: Optional[int], + last_update_from: Optional[datetime], + is_edited: Optional[bool], + ) -> Dict[str, Any]: """Build query parameters for GET requests.""" - params: dict[str, Any] = {} + params: Dict[str, Any] = {} if page is not None: params["page"] = page if per_page is not None: @@ -185,7 +185,7 @@ def _build_query_params( def _parse_paginated_response( self, - response: dict[str, Any], + response: Dict[str, Any], item_class: Type[T], ) -> PaginatedResponse[T]: """Parse paginated response into typed objects.""" @@ -203,8 +203,8 @@ def _request( method: str, endpoint: str, data: Any = None, - params: dict[str, Any] | None = None, - ) -> dict[str, Any]: + params: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: """Make HTTP request to the API.""" url = f"{self.base_url}{endpoint}" @@ -221,7 +221,7 @@ def _request( return self._handle_response(response) - def _handle_response(self, response: requests.Response) -> dict[str, Any]: + def _handle_response(self, response: requests.Response) -> Dict[str, Any]: """Handle API response.""" try: body = response.json() if response.text else {} diff --git a/src/pobo/dto/__init__.py b/src/pobo/dto/__init__.py index 9f3a0af..04445db 100644 --- a/src/pobo/dto/__init__.py +++ b/src/pobo/dto/__init__.py @@ -1,13 +1,13 @@ """DTO classes for Pobo SDK.""" -from pobo.dto.localized_string import LocalizedString -from pobo.dto.content import Content -from pobo.dto.product import Product -from pobo.dto.category import Category from pobo.dto.blog import Blog -from pobo.dto.parameter import Parameter, ParameterValue +from pobo.dto.category import Category +from pobo.dto.content import Content from pobo.dto.import_result import ImportResult +from pobo.dto.localized_string import LocalizedString from pobo.dto.paginated_response import PaginatedResponse +from pobo.dto.parameter import Parameter, ParameterValue +from pobo.dto.product import Product from pobo.dto.webhook_payload import WebhookPayload __all__ = [ diff --git a/src/pobo/dto/blog.py b/src/pobo/dto/blog.py index 88153ff..bef47da 100644 --- a/src/pobo/dto/blog.py +++ b/src/pobo/dto/blog.py @@ -3,12 +3,12 @@ from __future__ import annotations from datetime import datetime -from typing import Any +from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field -from pobo.dto.localized_string import LocalizedString from pobo.dto.content import Content +from pobo.dto.localized_string import LocalizedString class Blog(BaseModel): @@ -18,19 +18,19 @@ class Blog(BaseModel): is_visible: bool name: LocalizedString url: LocalizedString - category: str | None = None - description: LocalizedString | None = None - seo_title: LocalizedString | None = None - seo_description: LocalizedString | None = None - content: Content | None = None - images: list[str] = Field(default_factory=list) - is_loaded: bool | None = None - created_at: datetime | None = None - updated_at: datetime | None = None - - def to_api_dict(self) -> dict[str, Any]: + category: Optional[str] = None + description: Optional[LocalizedString] = None + seo_title: Optional[LocalizedString] = None + seo_description: Optional[LocalizedString] = None + content: Optional[Content] = None + images: List[str] = Field(default_factory=list) + is_loaded: Optional[bool] = None + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + def to_api_dict(self) -> Dict[str, Any]: """Convert to dictionary for API request.""" - data: dict[str, Any] = { + data: Dict[str, Any] = { "id": self.id, "is_visible": self.is_visible, "name": self.name.to_dict(), diff --git a/src/pobo/dto/category.py b/src/pobo/dto/category.py index 98f2d37..a439dc1 100644 --- a/src/pobo/dto/category.py +++ b/src/pobo/dto/category.py @@ -3,12 +3,12 @@ from __future__ import annotations from datetime import datetime -from typing import Any +from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field -from pobo.dto.localized_string import LocalizedString from pobo.dto.content import Content +from pobo.dto.localized_string import LocalizedString class Category(BaseModel): @@ -18,19 +18,19 @@ class Category(BaseModel): is_visible: bool name: LocalizedString url: LocalizedString - description: LocalizedString | None = None - seo_title: LocalizedString | None = None - seo_description: LocalizedString | None = None - content: Content | None = None - images: list[str] = Field(default_factory=list) - guid: str | None = None - is_loaded: bool | None = None - created_at: datetime | None = None - updated_at: datetime | None = None - - def to_api_dict(self) -> dict[str, Any]: + description: Optional[LocalizedString] = None + seo_title: Optional[LocalizedString] = None + seo_description: Optional[LocalizedString] = None + content: Optional[Content] = None + images: List[str] = Field(default_factory=list) + guid: Optional[str] = None + is_loaded: Optional[bool] = None + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + def to_api_dict(self) -> Dict[str, Any]: """Convert to dictionary for API request.""" - data: dict[str, Any] = { + data: Dict[str, Any] = { "id": self.id, "is_visible": self.is_visible, "name": self.name.to_dict(), diff --git a/src/pobo/dto/content.py b/src/pobo/dto/content.py index 68296c9..0eb3ff6 100644 --- a/src/pobo/dto/content.py +++ b/src/pobo/dto/content.py @@ -2,7 +2,9 @@ from __future__ import annotations -from pydantic import BaseModel +from typing import Dict, Optional, Union + +from pydantic import BaseModel, Field from pobo.enums import Language @@ -10,23 +12,31 @@ class Content(BaseModel): """HTML and marketplace content for multiple languages.""" - html: dict[str, str] = {} - marketplace: dict[str, str] = {} + html: Dict[str, str] = Field(default_factory=dict) + marketplace: Dict[str, str] = Field(default_factory=dict) + + def _get_language_key(self, language: Union[Language, str]) -> str: + """Get the string key for a language.""" + if isinstance(language, Language): + return language.value + return language - def get_html(self, language: Language | str) -> str | None: + def get_html(self, language: Union[Language, str]) -> Optional[str]: """Get HTML content for a specific language.""" - return self.html.get(str(language)) + key = self._get_language_key(language) + return self.html.get(key) - def get_marketplace(self, language: Language | str) -> str | None: + def get_marketplace(self, language: Union[Language, str]) -> Optional[str]: """Get marketplace content for a specific language.""" - return self.marketplace.get(str(language)) + key = self._get_language_key(language) + return self.marketplace.get(key) @property - def html_default(self) -> str | None: + def html_default(self) -> Optional[str]: """Get default HTML content.""" return self.html.get("default") @property - def marketplace_default(self) -> str | None: + def marketplace_default(self) -> Optional[str]: """Get default marketplace content.""" return self.marketplace.get("default") diff --git a/src/pobo/dto/import_result.py b/src/pobo/dto/import_result.py index 0cd1752..5b1e0c0 100644 --- a/src/pobo/dto/import_result.py +++ b/src/pobo/dto/import_result.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field @@ -14,9 +14,9 @@ class ImportResult(BaseModel): imported: int = 0 updated: int = 0 skipped: int = 0 - errors: list[dict[str, Any]] = Field(default_factory=list) - values_imported: int | None = None - values_updated: int | None = None + errors: List[Dict[str, Any]] = Field(default_factory=list) + values_imported: Optional[int] = None + values_updated: Optional[int] = None def has_errors(self) -> bool: """Check if there are any errors.""" diff --git a/src/pobo/dto/localized_string.py b/src/pobo/dto/localized_string.py index ff11c82..0115f39 100644 --- a/src/pobo/dto/localized_string.py +++ b/src/pobo/dto/localized_string.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Dict, Optional, Union + from pydantic import BaseModel, ConfigDict from pobo.enums import Language @@ -12,29 +14,37 @@ class LocalizedString(BaseModel): model_config = ConfigDict(extra="allow") - default: str | None = None - cs: str | None = None - sk: str | None = None - en: str | None = None - de: str | None = None - pl: str | None = None - hu: str | None = None + default: Optional[str] = None + cs: Optional[str] = None + sk: Optional[str] = None + en: Optional[str] = None + de: Optional[str] = None + pl: Optional[str] = None + hu: Optional[str] = None @classmethod def create(cls, default_value: str) -> LocalizedString: """Create a LocalizedString with a default value.""" return cls(default=default_value) - def with_translation(self, language: Language | str, value: str) -> LocalizedString: + def _get_language_key(self, language: Union[Language, str]) -> str: + """Get the string key for a language.""" + if isinstance(language, Language): + return language.value + return language + + def with_translation(self, language: Union[Language, str], value: str) -> LocalizedString: """Return a new LocalizedString with an additional translation.""" data = self.model_dump(exclude_none=True) - data[str(language)] = value + key = self._get_language_key(language) + data[key] = value return LocalizedString.model_validate(data) - def get(self, language: Language | str) -> str | None: + def get(self, language: Union[Language, str]) -> Optional[str]: """Get translation for a specific language.""" - return getattr(self, str(language), None) + key = self._get_language_key(language) + return getattr(self, key, None) - def to_dict(self) -> dict[str, str]: + def to_dict(self) -> Dict[str, str]: """Convert to dictionary, excluding None values.""" return self.model_dump(exclude_none=True) diff --git a/src/pobo/dto/paginated_response.py b/src/pobo/dto/paginated_response.py index a1b6cee..03777b4 100644 --- a/src/pobo/dto/paginated_response.py +++ b/src/pobo/dto/paginated_response.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Generic, TypeVar +from typing import Generic, List, TypeVar from pydantic import BaseModel, Field @@ -12,7 +12,7 @@ class PaginatedResponse(BaseModel, Generic[T]): """Paginated API response.""" - data: list[T] = Field(default_factory=list) + data: List[T] = Field(default_factory=list) current_page: int = 1 per_page: int = 100 total: int = 0 diff --git a/src/pobo/dto/parameter.py b/src/pobo/dto/parameter.py index a6dd9b5..77180a2 100644 --- a/src/pobo/dto/parameter.py +++ b/src/pobo/dto/parameter.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, Dict, List from pydantic import BaseModel, Field @@ -13,7 +13,7 @@ class ParameterValue(BaseModel): id: int value: str - def to_api_dict(self) -> dict[str, Any]: + def to_api_dict(self) -> Dict[str, Any]: """Convert to dictionary for API request.""" return {"id": self.id, "value": self.value} @@ -23,9 +23,9 @@ class Parameter(BaseModel): id: int name: str - values: list[ParameterValue] = Field(default_factory=list) + values: List[ParameterValue] = Field(default_factory=list) - def to_api_dict(self) -> dict[str, Any]: + def to_api_dict(self) -> Dict[str, Any]: """Convert to dictionary for API request.""" return { "id": self.id, diff --git a/src/pobo/dto/product.py b/src/pobo/dto/product.py index a447805..ecd94c0 100644 --- a/src/pobo/dto/product.py +++ b/src/pobo/dto/product.py @@ -3,12 +3,12 @@ from __future__ import annotations from datetime import datetime -from typing import Any +from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field -from pobo.dto.localized_string import LocalizedString from pobo.dto.content import Content +from pobo.dto.localized_string import LocalizedString class Product(BaseModel): @@ -18,23 +18,23 @@ class Product(BaseModel): is_visible: bool name: LocalizedString url: LocalizedString - short_description: LocalizedString | None = None - description: LocalizedString | None = None - seo_title: LocalizedString | None = None - seo_description: LocalizedString | None = None - content: Content | None = None - images: list[str] = Field(default_factory=list) - categories_ids: list[str] = Field(default_factory=list) - parameters_ids: list[int] = Field(default_factory=list) - guid: str | None = None - is_loaded: bool | None = None - categories: list[dict[str, Any]] = Field(default_factory=list) - created_at: datetime | None = None - updated_at: datetime | None = None - - def to_api_dict(self) -> dict[str, Any]: + short_description: Optional[LocalizedString] = None + description: Optional[LocalizedString] = None + seo_title: Optional[LocalizedString] = None + seo_description: Optional[LocalizedString] = None + content: Optional[Content] = None + images: List[str] = Field(default_factory=list) + categories_ids: List[str] = Field(default_factory=list) + parameters_ids: List[int] = Field(default_factory=list) + guid: Optional[str] = None + is_loaded: Optional[bool] = None + categories: List[Dict[str, Any]] = Field(default_factory=list) + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + def to_api_dict(self) -> Dict[str, Any]: """Convert to dictionary for API request.""" - data: dict[str, Any] = { + data: Dict[str, Any] = { "id": self.id, "is_visible": self.is_visible, "name": self.name.to_dict(), diff --git a/src/pobo/dto/webhook_payload.py b/src/pobo/dto/webhook_payload.py index 0529196..f953adc 100644 --- a/src/pobo/dto/webhook_payload.py +++ b/src/pobo/dto/webhook_payload.py @@ -3,6 +3,7 @@ from __future__ import annotations from datetime import datetime +from typing import Union from pydantic import BaseModel, field_validator @@ -16,7 +17,7 @@ class WebhookPayload(BaseModel): @field_validator("timestamp", mode="before") @classmethod - def parse_timestamp(cls, v: int | str | datetime) -> datetime: + def parse_timestamp(cls, v: Union[int, str, datetime]) -> datetime: """Parse timestamp from various formats.""" if isinstance(v, datetime): return v diff --git a/src/pobo/enums.py b/src/pobo/enums.py index c63baa1..f695b94 100644 --- a/src/pobo/enums.py +++ b/src/pobo/enums.py @@ -1,9 +1,9 @@ """Enums for Pobo SDK.""" -from enum import StrEnum +from enum import Enum -class Language(StrEnum): +class Language(str, Enum): """Supported languages for localized content.""" DEFAULT = "default" @@ -15,7 +15,7 @@ class Language(StrEnum): HU = "hu" -class WebhookEvent(StrEnum): +class WebhookEvent(str, Enum): """Webhook event types.""" PRODUCTS_UPDATE = "products.update" diff --git a/src/pobo/exceptions.py b/src/pobo/exceptions.py index 9fd7523..8b2b9a6 100644 --- a/src/pobo/exceptions.py +++ b/src/pobo/exceptions.py @@ -1,6 +1,6 @@ """Exceptions for Pobo SDK.""" -from typing import Any +from typing import Any, Dict, Optional class PoboError(Exception): @@ -15,7 +15,7 @@ class ApiError(PoboError): def __init__( self, message: str, - http_code: int | None = None, + http_code: Optional[int] = None, response_body: Any = None, ) -> None: super().__init__(message) @@ -38,7 +38,7 @@ def from_response(cls, http_code: int, body: Any) -> "ApiError": class ValidationError(PoboError): """Validation error with field errors.""" - def __init__(self, message: str, errors: dict[str, Any] | None = None) -> None: + def __init__(self, message: str, errors: Optional[Dict[str, Any]] = None) -> None: super().__init__(message) self.errors = errors or {} diff --git a/src/pobo/webhook_handler.py b/src/pobo/webhook_handler.py index 498c975..35cfa05 100644 --- a/src/pobo/webhook_handler.py +++ b/src/pobo/webhook_handler.py @@ -5,7 +5,7 @@ import hashlib import hmac import json -from typing import Any +from typing import Any, Optional, Union from pobo.dto.webhook_payload import WebhookPayload from pobo.exceptions import WebhookError @@ -19,7 +19,7 @@ class WebhookHandler: def __init__(self, webhook_secret: str) -> None: self.webhook_secret = webhook_secret - def handle(self, payload: str | bytes, signature: str | None) -> WebhookPayload: + def handle(self, payload: Union[str, bytes], signature: Optional[str]) -> WebhookPayload: """ Validate and parse webhook payload.