|
2 | 2 | import logging |
3 | 3 | import warnings |
4 | 4 | from pathlib import Path |
5 | | -from typing import Any, Dict, List, Literal, Optional, Tuple, Union |
| 5 | +from typing import Any, Dict, List, Literal, Optional, Sequence, Tuple, Union |
6 | 6 |
|
7 | 7 | import requests |
8 | 8 |
|
@@ -214,6 +214,131 @@ def get_products( |
214 | 214 | ).json() |
215 | 215 |
|
216 | 216 |
|
| 217 | +class FolksonomyResource: |
| 218 | + def __init__(self, api_config: APIConfig): |
| 219 | + self.api_config: APIConfig = api_config |
| 220 | + self.base_url = URLBuilder.folksonomy(environment=self.api_config.environment) |
| 221 | + |
| 222 | + def get( |
| 223 | + self, |
| 224 | + code: str, |
| 225 | + owner: str | None = None, |
| 226 | + keys: Sequence | str | None = None, |
| 227 | + ) -> requests.Response: |
| 228 | + """Retrieve folksonomy tags for a product. |
| 229 | +
|
| 230 | + :param code: the product barcode |
| 231 | + :param owner: the tag owner; requires authentication as that user. |
| 232 | + Case-sensitive. Leave empty for public tags (default). |
| 233 | + :param keys: List of keys to filter by. Can be provided as either |
| 234 | + a Python sequence (list, set, tuple, ...) or as |
| 235 | + a comma-separated string. |
| 236 | + If None (the default), all keys are returned. |
| 237 | + :return: the API response""" |
| 238 | + params: dict[str, str] = dict() |
| 239 | + if owner: |
| 240 | + params["owner"] = owner |
| 241 | + if keys: |
| 242 | + if not isinstance(keys, str): |
| 243 | + keys = ",".join(keys) |
| 244 | + params["keys"] = keys |
| 245 | + return send_request( |
| 246 | + url=f"{self.base_url}/product/{code}", |
| 247 | + api_config=self.api_config, |
| 248 | + method="get", |
| 249 | + params=params, |
| 250 | + ) |
| 251 | + |
| 252 | + def add( |
| 253 | + self, |
| 254 | + code: str, |
| 255 | + key: str, |
| 256 | + value: str, |
| 257 | + version: Literal[1, None] = None, |
| 258 | + owner: str | None = None, |
| 259 | + ) -> requests.Response: |
| 260 | + """Add a folksonomy tag to a product. |
| 261 | +
|
| 262 | + :param code: the product barcode |
| 263 | + :param key: the key for the tag |
| 264 | + :param value: the value to set for the tag |
| 265 | + :param version: version of the tag. Should be None or 1. |
| 266 | + :param owner: the tag owner; requires authentication as that user. |
| 267 | + Case-sensitive. Leave empty for public tags (default). |
| 268 | + :return: the API response""" |
| 269 | + params: dict[str, str | int] = dict() |
| 270 | + params["code"] = code |
| 271 | + params["k"] = key |
| 272 | + params["v"] = value |
| 273 | + if version: |
| 274 | + params["version"] = version |
| 275 | + if owner: |
| 276 | + params["owner"] = owner |
| 277 | + return send_request( |
| 278 | + url=f"{self.base_url}/product", |
| 279 | + api_config=self.api_config, |
| 280 | + method="post", |
| 281 | + json=params, |
| 282 | + ) |
| 283 | + |
| 284 | + def update( |
| 285 | + self, |
| 286 | + code: str, |
| 287 | + key: str, |
| 288 | + value: str, |
| 289 | + version: int, |
| 290 | + owner: str | None = None, |
| 291 | + ) -> requests.Response: |
| 292 | + """Update a folksonomy tag on a product. |
| 293 | +
|
| 294 | + :param code: the product barcode |
| 295 | + :param key: the key for the tag |
| 296 | + :param value: the value to set for the tag |
| 297 | + :param version: must be equal to previous version + 1 |
| 298 | + :param owner: the tag owner; requires authentication as that user. |
| 299 | + Case-sensitive. Leave empty for public tags (default). |
| 300 | + :return: the API response""" |
| 301 | + params: dict[str, str | int] = dict() |
| 302 | + params["code"] = code |
| 303 | + params["k"] = key |
| 304 | + params["v"] = value |
| 305 | + params["version"] = version |
| 306 | + if owner: |
| 307 | + params["owner"] = owner |
| 308 | + return send_request( |
| 309 | + url=f"{self.base_url}/product", |
| 310 | + api_config=self.api_config, |
| 311 | + method="put", |
| 312 | + json=params, |
| 313 | + ) |
| 314 | + |
| 315 | + def delete( |
| 316 | + self, |
| 317 | + code: str, |
| 318 | + key: str, |
| 319 | + version: int, |
| 320 | + owner: str | None = None, |
| 321 | + ) -> requests.Response: |
| 322 | + """Delete a folksonomy tag on a product. |
| 323 | +
|
| 324 | + :param code: the product barcode |
| 325 | + :param key: the key to delete |
| 326 | + :param version: the version the tag is at |
| 327 | + :param owner: the tag owner; requires authentication as that user. |
| 328 | + Case-sensitive. Leave empty for public tags (default). |
| 329 | + :return: the API response""" |
| 330 | + params: dict[str, str | int] = dict() |
| 331 | + params["version"] = version |
| 332 | + if owner: |
| 333 | + params["owner"] = owner |
| 334 | + return send_request( |
| 335 | + url=f"{self.base_url}/product/{code}/{key}", |
| 336 | + api_config=self.api_config, |
| 337 | + method="delete", |
| 338 | + params=params, |
| 339 | + ) |
| 340 | + |
| 341 | + |
217 | 342 | class ProductResource: |
218 | 343 | def __init__(self, api_config: APIConfig): |
219 | 344 | self.api_config = api_config |
@@ -636,6 +761,7 @@ def __init__( |
636 | 761 | self.country = country |
637 | 762 | self.product = ProductResource(self.api_config) |
638 | 763 | self.facet = FacetResource(self.api_config) |
| 764 | + self.folksonomy = FolksonomyResource(self.api_config) |
639 | 765 | self.robotoff = RobotoffResource(self.api_config) |
640 | 766 |
|
641 | 767 |
|
|
0 commit comments