Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions LICENSE
Original file line number Diff line number Diff line change
Expand Up @@ -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.

1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ dev = [
"responses>=0.23.0",
"ruff>=0.1.0",
"mypy>=1.0.0",
"types-requests>=2.28.0",
]

[project.urls]
Expand Down
18 changes: 9 additions & 9 deletions src/pobo/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down
72 changes: 36 additions & 36 deletions src/pobo/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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]
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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
Expand All @@ -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()
Expand All @@ -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:
Expand All @@ -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."""
Expand All @@ -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}"

Expand All @@ -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 {}
Expand Down
10 changes: 5 additions & 5 deletions src/pobo/dto/__init__.py
Original file line number Diff line number Diff line change
@@ -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__ = [
Expand Down
28 changes: 14 additions & 14 deletions src/pobo/dto/blog.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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(),
Expand Down
28 changes: 14 additions & 14 deletions src/pobo/dto/category.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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(),
Expand Down
28 changes: 19 additions & 9 deletions src/pobo/dto/content.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,31 +2,41 @@

from __future__ import annotations

from pydantic import BaseModel
from typing import Dict, Optional, Union

from pydantic import BaseModel, Field

from pobo.enums import Language


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")
Loading