vinted_scraper

 31@dataclass
 32class AsyncVintedWrapper(BaseVintedWrapper):
 33    """Asynchronous Vinted API wrapper returning raw JSON responses.
 34
 35    Handles cookie management, retries, and async HTTP requests automatically.
 36    Returns raw JSON dictionaries instead of typed objects.
 37
 38    Attributes:
 39        baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 40        session_cookie: Session cookie dict. Auto-fetched if None.
 41        user_agent: Custom user agent string. Auto-generated if None.
 42        config: httpx client configuration dict.
 43        cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 44
 45    Example:
 46        See https://github.com/Giglium/vinted_scraper/blob/main/examples/async_wrapper.py
 47    """
 48
 49    _client: httpx.AsyncClient = field(init=False, repr=False)
 50
 51    @classmethod
 52    async def create(
 53        cls,
 54        baseurl: str,
 55        user_agent: Optional[str] = None,
 56        config: Optional[Dict] = None,
 57        cookie_names: Optional[List[str]] = None,
 58    ):
 59        """Factory method to create an AsyncVintedWrapper instance.
 60
 61        Use this instead of direct instantiation to automatically fetch the session cookie.
 62
 63        Args:
 64            baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 65            user_agent: Custom user agent string. Auto-generated if None.
 66            config: httpx client configuration dict.
 67            cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 68
 69        Returns:
 70            Initialized AsyncVintedWrapper instance with fetched cookies.
 71        """
 72        _log.debug("Creating the async wrapper using the factory method")
 73        self = cls(
 74            baseurl, user_agent=user_agent, config=config, cookie_names=cookie_names
 75        )
 76        self.session_cookie = await self.refresh_cookie()
 77        return self
 78
 79    def __post_init__(self) -> None:
 80        """Initialize AsyncVintedWrapper after dataclass initialization.
 81
 82        Validates the base URL, sets up user agent, and initializes httpx async client.
 83
 84        Raises:
 85            RuntimeError: If the base URL is invalid.
 86
 87        Note:
 88            Use the create() factory method instead of direct instantiation to
 89            automatically fetch the session cookie.
 90        """
 91        httpx_config = self._validate_and_init()
 92        self._client = httpx.AsyncClient(**httpx_config)
 93
 94    async def refresh_cookie(self, retries: int = DEFAULT_RETRIES) -> Dict[str, str]:
 95        """Manually refresh the session cookie asynchronously.
 96
 97        Args:
 98            retries: Number of retry attempts (default: 3).
 99
100        Returns:
101            Dictionary containing session cookies.
102
103        Raises:
104            RuntimeError: If cookies cannot be fetched after all retries.
105        """
106        log_refresh_cookie(_log)
107        return await AsyncVintedWrapper.fetch_cookie(
108            self._client,
109            self._get_cookie_headers(),
110            self.cookie_names,
111            retries,
112        )
113
114    @staticmethod
115    async def fetch_cookie(
116        client: httpx.AsyncClient,
117        headers: Dict,
118        cookie_names: List[str],
119        retries: int = DEFAULT_RETRIES,
120    ) -> Dict[str, str]:
121        """Fetch session cookies from Vinted using async HTTP GET request.
122
123        Args:
124            client: httpx.AsyncClient instance.
125            headers: HTTP headers dictionary.
126            cookie_names: List of cookie names to extract.
127            retries: Number of retry attempts (default: 3).
128
129        Returns:
130            Dictionary of extracted session cookies.
131
132        Raises:
133            RuntimeError: If cookies cannot be fetched after all retries.
134        """
135        response = None
136
137        for i in range(retries):
138            log_interaction(_log, i, retries)
139            response = await client.get("/", headers=headers)
140
141            cookies = BaseVintedWrapper._process_cookie_response(response, cookie_names)
142            if cookies:
143                return cookies
144
145            if response.status_code != HTTP_OK:
146                sleep_time = BaseVintedWrapper._handle_cookie_failure(
147                    response, i, retries
148                )
149                if i < retries - 1:
150                    await asyncio.sleep(sleep_time)
151
152        BaseVintedWrapper._raise_cookie_error(client.base_url, response)
153
154    async def search(self, params: Optional[Dict] = None) -> Dict[str, Any]:
155        """Search for items on Vinted asynchronously.
156
157        Args:
158            params: Query parameters. Common parameters:
159                - search_text: Search query
160                - page: Page number
161                - per_page: Items per page
162                - price_from: Minimum price
163                - price_to: Maximum price
164                - order: Sort order
165                - catalog_ids: Category IDs
166                - brand_ids: Brand IDs
167                - size_ids: Size IDs
168
169        Returns:
170            Dictionary containing JSON response with search results.
171        """
172        log_search(_log, params)
173        return await self.curl(self._search_endpoint(), params=params)
174
175    async def item(
176        self, item_id: str, fields: Optional[List[str]] = None
177    ) -> Dict[str, Any]:
178        """Read item metadata from the public item page (HTML), asynchronously.
179
180        The JSON item endpoint (``/api/v2/items/{id}/details``) is blocked by the
181        anti-bot protection and returns ``403`` (see
182        https://github.com/Giglium/vinted_scraper/issues/59), so the item data is
183        read from the public item page instead. Uses HTTP streaming to download
184        only the ``<head>`` section, extracting OpenGraph meta tags without
185        fetching the full page body.
186
187        Args:
188            item_id: The unique identifier of the item.
189            fields: List of ``OgField`` values to extract. Defaults to all
190                fields (``[OgField.TITLE, OgField.DESCRIPTION, OgField.URL,
191                OgField.IMAGE]``).
192
193        Returns:
194            A dict always containing ``id``, plus keys ``title``,
195            ``description``, ``url``, and ``image`` (each present only if
196            found and requested).
197
198        Raises:
199            RuntimeError: If the item page cannot be fetched (non-200 status).
200        """
201        log_item(_log, item_id, fields)
202        endpoint = self._item_endpoint(item_id)
203        headers = self._build_page_headers()
204
205        parts: List[str] = []
206        async with self._client.stream("GET", endpoint, headers=headers) as response:
207            status_code = response.status_code
208            if status_code == HTTP_OK:
209                tail = ""
210                async for chunk in response.aiter_text(chunk_size=4096):
211                    parts.append(chunk)
212                    # Check boundary: </head> may span two consecutive chunks
213                    combined = tail + chunk.lower()
214                    if "</head>" in combined:
215                        break
216                    tail = chunk[-6:].lower()
217            else:
218                await response.aread()
219        head_html = "".join(parts)
220
221        log_curl_response(_log, endpoint, status_code, response.headers, head_html)
222
223        if status_code == HTTP_OK:
224            return parse_item_page(item_id, head_html, fields)
225
226        self._raise_curl_error(endpoint, status_code)
227
228    async def curl(
229        self,
230        endpoint: str,
231        params: Optional[Dict] = None,
232        *,
233        _retries: int = 0,
234    ) -> Dict[str, Any]:
235        """Send an async HTTP GET request to any Vinted API endpoint.
236
237        Automatically handles headers, cookies, retries, and error responses.
238
239        Args:
240            endpoint: API endpoint path (e.g., "/api/v2/users/username").
241            params: Optional query parameters.
242
243        Returns:
244            Dictionary containing the parsed JSON response.
245
246        Raises:
247            RuntimeError: If response status is not 200 or JSON parsing fails.
248        """
249        headers = self._build_curl_headers()
250        log_curl_request(_log, self.baseurl, endpoint, headers, params)
251
252        response = await self._client.get(endpoint, headers=headers, params=params)
253
254        log_curl_response(
255            _log, endpoint, response.status_code, response.headers, response.text
256        )
257
258        if response.status_code == HTTP_OK:
259            return self._handle_curl_response(response, endpoint)
260
261        if response.status_code == HTTP_UNAUTHORIZED and _retries < DEFAULT_RETRIES:
262            log_cookie_retry(_log, response.status_code)
263            self.session_cookie = await self.refresh_cookie()
264            return await self.curl(endpoint, params, _retries=_retries + 1)
265
266        self._raise_curl_error(endpoint, response.status_code)
267
268    async def __aenter__(self) -> "AsyncVintedWrapper":  # pragma: no cover
269        """Enter async context manager.
270
271        Returns:
272            Self for use in async with statement.
273        """
274        return self
275
276    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:  # pragma: no cover
277        """Exit async context manager and close HTTP client.
278
279        Args:
280            exc_type: Exception type (unused).
281            exc_val: Exception value (unused).
282            exc_tb: Exception traceback (unused).
283        """
284        await self._client.aclose()
285
286    def __del__(self) -> None:  # pragma: no cover
287        """Best-effort cleanup of the HTTP client on garbage collection.
288
289        Prefer using the async context manager (``async with`` statement)
290        for deterministic resource cleanup.
291
292        Note: httpx.AsyncClient exposes ``aclose()`` (async) but not
293        ``close()`` (sync). Since ``__del__`` cannot await, we attempt a
294        synchronous close via the underlying transport if available.
295        """
296        if hasattr(self, "_client") and not self._client.is_closed:
297            try:
298                # httpx >=0.28 removed the sync close() helper on AsyncClient
299                self._client.close()  # type: ignore[attr-defined]
300            except AttributeError:
301                # Fallback: close the underlying transport directly
302                transport = getattr(self._client, "_transport", None)
303                if transport is not None and hasattr(transport, "close"):
304                    transport.close()

 31@dataclass
 32class VintedWrapper(BaseVintedWrapper):
 33    """Synchronous Vinted API wrapper returning raw JSON responses.
 34
 35    Handles cookie management, retries, and HTTP requests automatically.
 36    Returns raw JSON dictionaries instead of typed objects.
 37
 38    Attributes:
 39        baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 40        session_cookie: Session cookie dict. Auto-fetched if None.
 41        user_agent: Custom user agent string. Auto-generated if None.
 42        config: httpx client configuration dict.
 43        cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 44
 45    Example:
 46        See https://github.com/Giglium/vinted_scraper/blob/main/examples/wrapper.py
 47    """
 48
 49    _client: httpx.Client = field(init=False, repr=False)
 50
 51    def __post_init__(self) -> None:
 52        """Initialize VintedWrapper after dataclass initialization.
 53
 54        Validates the base URL, sets up user agent, initializes httpx client,
 55        and fetches session cookies if not provided.
 56
 57        Raises:
 58            RuntimeError: If the base URL is invalid.
 59        """
 60        httpx_config = self._validate_and_init()
 61        self._client = httpx.Client(**httpx_config)
 62        if self.session_cookie is None:
 63            self.session_cookie = self.refresh_cookie()
 64
 65    def refresh_cookie(self, retries: int = DEFAULT_RETRIES) -> Dict[str, str]:
 66        """Manually refresh the session cookie.
 67
 68        Args:
 69            retries: Number of retry attempts (default: 3).
 70
 71        Returns:
 72            Dictionary containing session cookies.
 73
 74        Raises:
 75            RuntimeError: If cookies cannot be fetched after all retries.
 76        """
 77        log_refresh_cookie(_log)
 78        return VintedWrapper.fetch_cookie(
 79            self._client,
 80            self._get_cookie_headers(),
 81            self.cookie_names,
 82            retries,
 83        )
 84
 85    @staticmethod
 86    def fetch_cookie(
 87        client: httpx.Client,
 88        headers: Dict,
 89        cookie_names: List[str],
 90        retries: int = DEFAULT_RETRIES,
 91    ) -> Dict[str, str]:
 92        """Fetch session cookies from Vinted using HTTP GET request.
 93
 94        Args:
 95            client: httpx.Client instance.
 96            headers: HTTP headers dictionary.
 97            cookie_names: List of cookie names to extract.
 98            retries: Number of retry attempts (default: 3).
 99
100        Returns:
101            Dictionary of extracted session cookies.
102
103        Raises:
104            RuntimeError: If cookies cannot be fetched after all retries.
105        """
106        response = None
107
108        for i in range(retries):
109            log_interaction(_log, i, retries)
110            response = client.get("/", headers=headers)
111
112            cookies = BaseVintedWrapper._process_cookie_response(response, cookie_names)
113            if cookies:
114                return cookies
115
116            if response.status_code != HTTP_OK:
117                sleep_time = BaseVintedWrapper._handle_cookie_failure(
118                    response, i, retries
119                )
120                if i < retries - 1:
121                    time.sleep(sleep_time)
122
123        BaseVintedWrapper._raise_cookie_error(client.base_url, response)
124
125    def search(self, params: Optional[Dict] = None) -> Dict[str, Any]:
126        """Search for items on Vinted.
127
128        Args:
129            params: Query parameters. Common parameters:
130                - search_text (str): Search query
131                - page (int): Page number
132                - per_page (int): Items per page
133                - price_from (float): Minimum price
134                - price_to (float): Maximum price
135                - order (str): Sort order
136                - catalog_ids (str): Category IDs
137                - brand_ids (str): Brand IDs
138                - size_ids (str): Size IDs
139
140        Returns:
141            Dictionary containing JSON response with search results.
142        """
143        log_search(_log, params)
144        return self.curl(self._search_endpoint(), params=params)
145
146    def item(self, item_id: str, fields: Optional[List[str]] = None) -> Dict[str, Any]:
147        """Read item metadata from the public item page (HTML).
148
149        The JSON item endpoint (``/api/v2/items/{id}/details``) is blocked by the
150        anti-bot protection and returns ``403`` (see
151        https://github.com/Giglium/vinted_scraper/issues/59), so the item data is
152        read from the public item page instead. Uses HTTP streaming to download
153        only the ``<head>`` section, extracting OpenGraph meta tags without
154        fetching the full page body.
155
156        Args:
157            item_id: The unique identifier of the item.
158            fields: List of ``OgField`` values to extract. Defaults to all
159                fields (``[OgField.TITLE, OgField.DESCRIPTION, OgField.URL,
160                OgField.IMAGE]``).
161
162        Returns:
163            A dict always containing ``id``, plus keys ``title``,
164            ``description``, ``url``, and ``image`` (each present only if
165            found and requested).
166
167        Raises:
168            RuntimeError: If the item page cannot be fetched (non-200 status).
169        """
170        log_item(_log, item_id, fields)
171        endpoint = self._item_endpoint(item_id)
172        headers = self._build_page_headers()
173
174        parts: List[str] = []
175        with self._client.stream("GET", endpoint, headers=headers) as response:
176            status_code = response.status_code
177            if status_code == HTTP_OK:
178                tail = ""
179                for chunk in response.iter_text(chunk_size=4096):
180                    parts.append(chunk)
181                    # Check boundary: </head> may span two consecutive chunks
182                    combined = tail + chunk.lower()
183                    if "</head>" in combined:
184                        break
185                    tail = chunk[-6:].lower()
186            else:
187                response.read()
188        head_html = "".join(parts)
189
190        log_curl_response(_log, endpoint, status_code, response.headers, head_html)
191
192        if status_code == HTTP_OK:
193            return parse_item_page(item_id, head_html, fields)
194
195        self._raise_curl_error(endpoint, status_code)
196
197    def curl(
198        self,
199        endpoint: str,
200        params: Optional[Dict] = None,
201        *,
202        _retries: int = 0,
203    ) -> Dict[str, Any]:
204        """Send a custom HTTP GET request to any Vinted API endpoint.
205
206        Automatically handles headers, cookies, retries, and error responses.
207
208        Args:
209            endpoint: API endpoint path (e.g., "/api/v2/users/username").
210            params: Optional query parameters.
211
212        Returns:
213            Dictionary containing the parsed JSON response.
214
215        Raises:
216            RuntimeError: If response status is not 200 or JSON parsing fails.
217        """
218        headers = self._build_curl_headers()
219        log_curl_request(_log, self.baseurl, endpoint, headers, params)
220
221        response = self._client.get(endpoint, headers=headers, params=params)
222
223        log_curl_response(
224            _log, endpoint, response.status_code, response.headers, response.text
225        )
226
227        if response.status_code == HTTP_OK:
228            return self._handle_curl_response(response, endpoint)
229
230        if response.status_code == HTTP_UNAUTHORIZED and _retries < DEFAULT_RETRIES:
231            log_cookie_retry(_log, response.status_code)
232            self.session_cookie = self.refresh_cookie()
233            return self.curl(endpoint, params, _retries=_retries + 1)
234
235        self._raise_curl_error(endpoint, response.status_code)
236
237    def __enter__(self) -> "VintedWrapper":
238        """Enter context manager.
239
240        Returns:
241            Self for use in with statement.
242        """
243        return self
244
245    def __exit__(self, exc_type, exc_val, exc_tb) -> None:  # pragma: no cover
246        """Exit context manager and close HTTP client.
247
248        Args:
249            exc_type: Exception type (unused).
250            exc_val: Exception value (unused).
251            exc_tb: Exception traceback (unused).
252        """
253        self._client.close()
254
255    def __del__(self) -> None:  # pragma: no cover
256        """Best-effort cleanup of the HTTP client on garbage collection.
257
258        Prefer using the context manager (``with`` statement) for
259        deterministic resource cleanup.
260        """
261        if hasattr(self, "_client") and not self._client.is_closed:
262            self._client.close()

 14@dataclass
 15class AsyncVintedScraper(AsyncVintedWrapper):
 16    """Asynchronous Vinted scraper with typed model support.
 17
 18    Returns structured VintedItem objects instead of raw JSON dictionaries.
 19    Inherits all functionality from AsyncVintedWrapper.
 20
 21    Example:
 22        See https://github.com/Giglium/vinted_scraper/blob/main/examples/async_scraper.py
 23    """
 24
 25    async def search(self, params: Optional[Dict] = None) -> List[VintedItem]:
 26        """Search for items on Vinted asynchronously.
 27
 28        Args:
 29            params: Query parameters for the search. Common parameters:
 30                - search_text: Search query
 31                - page: Page number
 32                - per_page: Items per page
 33                - price_from: Minimum price
 34                - price_to: Maximum price
 35                - order: Sort order
 36                - catalog_ids: Category IDs
 37                - brand_ids: Brand IDs
 38                - size_ids: Size IDs
 39
 40        Returns:
 41            List of VintedItem objects representing search results.
 42        """
 43        response = await super().search(params)
 44        return [VintedItem(json_data=item) for item in response["items"]]
 45
 46    async def item(
 47        self, item_id: str, fields: Optional[List[str]] = None
 48    ) -> VintedItem:
 49        """Read item metadata from the public item page (HTML), asynchronously.
 50
 51        The JSON item endpoint is blocked by the anti-bot protection and returns
 52        ``403`` (see https://github.com/Giglium/vinted_scraper/issues/59), so the
 53        data is read from the public item page instead. Only the fields exposed
 54        by the page's OpenGraph tags are populated (``title``, ``description``,
 55        ``url``, ``image``).
 56
 57        Args:
 58            item_id: The unique identifier of the item.
 59            fields: List of ``OgField`` values to extract. Defaults to all
 60                fields (``[OgField.TITLE, OgField.DESCRIPTION, OgField.URL,
 61                OgField.IMAGE]``).
 62
 63        Returns:
 64            A VintedItem built from the page metadata. Always contains ``id``.
 65
 66        Raises:
 67            RuntimeError: If the item page cannot be fetched (non-200 status).
 68        """
 69        data = await super().item(item_id, fields)
 70        return VintedItem(json_data=data)
 71
 72    async def enrich(self, item: VintedItem) -> VintedItem:
 73        """Enrich an existing VintedItem with description from the item page.
 74
 75        Fetches the ``og:description`` from the public item page and populates
 76        the item's ``description`` attribute. Useful for enriching items
 77        obtained from search results (which lack a full description).
 78
 79        Args:
 80            item: A VintedItem to enrich (must have a valid ``id``).
 81
 82        Returns:
 83            The same VintedItem instance with ``description`` populated.
 84
 85        Raises:
 86            RuntimeError: If the item page cannot be fetched (non-200 status).
 87        """
 88        data = await super().item(str(item.id), [OgField.DESCRIPTION])
 89        if OgField.DESCRIPTION in data:
 90            item.description = data[OgField.DESCRIPTION]
 91        return item
 92
 93    async def curl(
 94        self, endpoint: str, params: Optional[Dict] = None, *, _retries: int = 0
 95    ) -> VintedJsonModel:
 96        """Send an async HTTP GET request to any Vinted API endpoint.
 97
 98        Args:
 99            endpoint: The API endpoint path (e.g., "/api/v2/users/username").
100            params: Optional query parameters.
101
102        Returns:
103            VintedJsonModel containing the JSON response.
104
105        Raises:
106            RuntimeError: If the request fails or returns a non-200 status code.
107        """
108        response = await super().curl(endpoint, params, _retries=_retries)
109        return VintedJsonModel(json_data=response)

 14@dataclass
 15class VintedScraper(VintedWrapper):
 16    """Synchronous Vinted scraper with typed model support.
 17
 18    Returns structured VintedItem objects instead of raw JSON dictionaries.
 19    Inherits all functionality from VintedWrapper.
 20
 21    Example:
 22       See https://github.com/Giglium/vinted_scraper/blob/main/examples/scraper.py
 23    """
 24
 25    def search(self, params: Optional[Dict] = None) -> List[VintedItem]:  # type: ignore
 26        """Search for items on Vinted.
 27
 28        Args:
 29            params: Query parameters for the search. Common parameters:
 30                - search_text: Search query
 31                - page: Page number
 32                - per_page: Items per page
 33                - price_from: Minimum price
 34                - price_to: Maximum price
 35                - order: Sort order
 36                - catalog_ids: Category IDs
 37                - brand_ids: Brand IDs
 38                - size_ids : Size IDs
 39
 40        Returns:
 41            List of VintedItem objects representing search results.
 42        """
 43        return [VintedItem(json_data=item) for item in super().search(params)["items"]]
 44
 45    def item(
 46        self, item_id: str, fields: Optional[List[str]] = None
 47    ) -> VintedItem:  # type: ignore
 48        """Read item metadata from the public item page (HTML).
 49
 50        The JSON item endpoint is blocked by the anti-bot protection and returns
 51        ``403`` (see https://github.com/Giglium/vinted_scraper/issues/59), so the
 52        data is read from the public item page instead. Only the fields exposed
 53        by the page's OpenGraph tags are populated (``title``, ``description``,
 54        ``url``, ``image``).
 55
 56        Args:
 57            item_id: The unique identifier of the item.
 58            fields: List of ``OgField`` values to extract. Defaults to all
 59                fields (``[OgField.TITLE, OgField.DESCRIPTION, OgField.URL,
 60                OgField.IMAGE]``).
 61
 62        Returns:
 63            A VintedItem built from the page metadata. Always contains ``id``.
 64
 65        Raises:
 66            RuntimeError: If the item page cannot be fetched (non-200 status).
 67        """
 68        data = super().item(item_id, fields)
 69        return VintedItem(json_data=data)
 70
 71    def enrich(self, item: VintedItem) -> VintedItem:
 72        """Enrich an existing VintedItem with description from the item page.
 73
 74        Fetches the ``og:description`` from the public item page and populates
 75        the item's ``description`` attribute. Useful for enriching items
 76        obtained from search results (which lack a full description).
 77
 78        Args:
 79            item: A VintedItem to enrich (must have a valid ``id``).
 80
 81        Returns:
 82            The same VintedItem instance with ``description`` populated.
 83
 84        Raises:
 85            RuntimeError: If the item page cannot be fetched (non-200 status).
 86        """
 87        data = super().item(str(item.id), [OgField.DESCRIPTION])
 88        if OgField.DESCRIPTION in data:
 89            item.description = data[OgField.DESCRIPTION]
 90        return item
 91
 92    def curl(
 93        self, endpoint: str, params: Optional[Dict] = None, *, _retries: int = 0
 94    ) -> VintedJsonModel:  # type: ignore
 95        """Send a custom HTTP GET request to any Vinted API endpoint.
 96
 97        Args:
 98            endpoint: The API endpoint path (e.g., "/api/v2/users/username").
 99            params: Optional query parameters.
100
101        Returns:
102            VintedJsonModel containing the JSON response.
103
104        Raises:
105            RuntimeError: If the request fails or returns a non-200 status code.
106        """
107        response = super().curl(endpoint, params, _retries=_retries)
108        return VintedJsonModel(json_data=response)