vinted_scraper

 43@dataclass
 44class AsyncVintedWrapper:
 45    """Asynchronous Vinted API wrapper returning raw JSON responses.
 46
 47    Handles cookie management, retries, and async HTTP requests automatically.
 48    Returns raw JSON dictionaries instead of typed objects.
 49
 50    Attributes:
 51        baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 52        session_cookie: Session cookie dict. Auto-fetched if None.
 53        user_agent: Custom user agent string. Auto-generated if None.
 54        config: httpx client configuration dict.
 55        cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 56
 57    Example:
 58        See https://github.com/Giglium/vinted_scraper/blob/main/examples/async_wrapper.py
 59    """
 60
 61    baseurl: str
 62    session_cookie: Optional[Dict[str, str]] = None
 63    user_agent: Optional[str] = None
 64    config: Optional[Dict] = None
 65    cookie_names: Optional[List[str]] = None
 66    _client: httpx.AsyncClient = field(init=False, repr=False)
 67
 68    @classmethod
 69    async def create(
 70        cls,
 71        baseurl: str,
 72        user_agent: Optional[str] = None,
 73        config: Optional[Dict] = None,
 74        cookie_names: Optional[List[str]] = None,
 75    ):
 76        """Factory method to create an AsyncVintedWrapper instance.
 77
 78        Use this instead of direct instantiation to automatically fetch the session cookie.
 79
 80        Args:
 81            baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 82            user_agent: Custom user agent string. Auto-generated if None.
 83            config: httpx client configuration dict.
 84            cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 85
 86        Returns:
 87            Initialized AsyncVintedWrapper instance with fetched cookies.
 88        """
 89        _log.debug("Creating the async wrapper using the factory method")
 90        self = cls(
 91            baseurl, user_agent=user_agent, config=config, cookie_names=cookie_names
 92        )
 93        self.session_cookie = await self.refresh_cookie()
 94        return self
 95
 96    def __post_init__(self) -> None:
 97        """Initialize AsyncVintedWrapper after dataclass initialization.
 98
 99        Validates the base URL, sets up user agent, and initializes httpx async client.
100
101        Raises:
102            RuntimeError: If the base URL is invalid.
103
104        Note:
105            Use the create() factory method instead of direct instantiation to
106            automatically fetch the session cookie.
107        """
108        if not url_validator(self.baseurl):
109            _log.error("'%s' is not a valid url", self.baseurl)
110            raise RuntimeError(f"'{self.baseurl}' is not a valid url, please check it!")
111
112        log_constructor(
113            log=_log,
114            self=self,
115            baseurl=self.baseurl,
116            user_agent=self.user_agent,
117            session_cookie=self.session_cookie,
118            config=self.config,
119        )
120
121        if self.user_agent is None:
122            self.user_agent = get_random_user_agent()
123        if self.cookie_names is None:
124            self.cookie_names = [SESSION_COOKIE_NAME]
125        self._client = httpx.AsyncClient(**get_httpx_config(self.baseurl, self.config))
126
127    async def refresh_cookie(self, retries: int = DEFAULT_RETRIES) -> Dict[str, str]:
128        """Manually refresh the session cookie asynchronously.
129
130        Args:
131            retries: Number of retry attempts (default: 3).
132
133        Returns:
134            Dictionary containing session cookies.
135
136        Raises:
137            RuntimeError: If cookies cannot be fetched after all retries.
138        """
139        log_refresh_cookie(_log)
140        return await AsyncVintedWrapper.fetch_cookie(
141            self._client,
142            get_cookie_headers(self.baseurl, self.user_agent),
143            self.cookie_names,
144            retries,
145        )
146
147    @staticmethod
148    async def fetch_cookie(
149        client: httpx.AsyncClient,
150        headers: Dict,
151        cookie_names: List[str],
152        retries: int = DEFAULT_RETRIES,
153    ) -> Dict[str, str]:
154        """Fetch session cookies from Vinted using async HTTP GET request.
155
156        Args:
157            client: httpx.AsyncClient instance.
158            headers: HTTP headers dictionary.
159            cookie_names: List of cookie names to extract.
160            retries: Number of retry attempts (default: 3).
161
162        Returns:
163            Dictionary of extracted session cookies.
164
165        Raises:
166            RuntimeError: If cookies cannot be fetched after all retries.
167        """
168        response = None
169
170        for i in range(retries):
171            log_interaction(_log, i, retries)
172            response = await client.get("/", headers=headers)
173
174            if response.status_code == HTTP_OK:
175                cookies = extract_cookie_from_response(response, cookie_names)
176                if cookies:
177                    log_cookie_fetched(_log, str(cookies))
178                    return cookies
179                _log.warning("Cannot find session cookie in response")
180            else:
181                log_cookie_fetch_failed(_log, response.status_code, i, retries)
182                sleep_time = RETRY_BASE_SLEEP**i
183                log_sleep(_log, sleep_time)
184                await asyncio.sleep(sleep_time)
185
186        _log.error("Cannot fetch session cookie from %s", client.base_url)
187        raise RuntimeError(
188            f"Cannot fetch session cookie from {client.base_url}, because of "
189            f"status code: {response.status_code if response is not None else 'none'}"
190            "different from 200."
191        )
192
193    async def search(self, params: Optional[Dict] = None) -> Dict[str, Any]:
194        """Search for items on Vinted asynchronously.
195
196        Args:
197            params: Query parameters. Common parameters:
198                - search_text: Search query
199                - page: Page number
200                - per_page: Items per page
201                - price_from: Minimum price
202                - price_to: Maximum price
203                - order: Sort order
204                - catalog_ids: Category IDs
205                - brand_ids: Brand IDs
206                - size_ids: Size IDs
207
208        Returns:
209            Dictionary containing JSON response with search results.
210        """
211        log_search(_log, params)
212        return await self.curl(API_CATALOG_ITEMS, params=params)
213
214    async def item(self, item_id: str, params: Optional[Dict] = None) -> Dict[str, Any]:
215        """Retrieve detailed information about a specific item asynchronously.
216
217        Args:
218            item_id: The unique identifier of the item.
219            params: Optional query parameters.
220
221        Returns:
222            Dictionary containing JSON response with item details.
223
224        Raises:
225            RuntimeError: If the item is not found or API returns an error.
226
227        Note:
228            It returns a 403 error after a few uses.
229            See: https://github.com/Giglium/vinted_scraper/issues/59
230        """
231        log_item(_log, item_id, params)
232        return await self.curl(f"{API_ITEMS}/{item_id}", params=params)
233
234    async def curl(
235        self, endpoint: str, params: Optional[Dict] = None
236    ) -> Dict[str, Any]:
237        """Send an async HTTP GET request to any Vinted API endpoint.
238
239        Automatically handles headers, cookies, retries, and error responses.
240
241        Args:
242            endpoint: API endpoint path (e.g., "/api/v2/users/username").
243            params: Optional query parameters.
244
245        Returns:
246            Dictionary containing the parsed JSON response.
247
248        Raises:
249            RuntimeError: If response status is not 200 or JSON parsing fails.
250        """
251        headers = get_curl_headers(self.baseurl, self.user_agent, self.session_cookie)
252
253        log_curl_request(_log, self.baseurl, endpoint, headers, params)
254
255        response = await self._client.get(
256            endpoint,
257            headers=get_curl_headers(
258                self.baseurl, self.user_agent, self.session_cookie
259            ),
260            params=params,
261        )
262
263        log_curl_response(
264            _log, endpoint, response.status_code, response.headers, response.text
265        )
266
267        # Success
268        if response.status_code == HTTP_OK:
269            try:
270                return response.json()
271            except ValueError as e:
272                _log.error("Failed to parse JSON response from %s: %s", endpoint, e)
273                raise RuntimeError(f"Invalid JSON response from {endpoint}: {e}") from e
274
275        # Fetch (maybe is expired?) the session cookie again and retry the API call
276        if response.status_code == HTTP_UNAUTHORIZED:
277            log_cookie_retry(_log, response.status_code)
278            self.session_cookie = await self.refresh_cookie()
279            return await self.curl(endpoint, params)
280        raise RuntimeError(
281            f"Cannot perform API call to endpoint {endpoint}, error code: {response.status_code}"
282        )
283
284    async def __aenter__(self) -> "AsyncVintedWrapper":  # pragma: no cover
285        """Enter async context manager.
286
287        Returns:
288            Self for use in async with statement.
289        """
290        return self
291
292    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:  # pragma: no cover
293        """Exit async context manager and close HTTP client.
294
295        Args:
296            exc_type: Exception type (unused).
297            exc_val: Exception value (unused).
298            exc_tb: Exception traceback (unused).
299        """
300        await self._client.aclose()

 43@dataclass
 44class VintedWrapper:
 45    """Synchronous Vinted API wrapper returning raw JSON responses.
 46
 47    Handles cookie management, retries, and HTTP requests automatically.
 48    Returns raw JSON dictionaries instead of typed objects.
 49
 50    Attributes:
 51        baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 52        session_cookie: Session cookie dict. Auto-fetched if None.
 53        user_agent: Custom user agent string. Auto-generated if None.
 54        config: httpx client configuration dict.
 55        cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 56
 57    Example:
 58        See https://github.com/Giglium/vinted_scraper/blob/main/examples/wrapper.py
 59    """
 60
 61    baseurl: str
 62    session_cookie: Optional[Dict[str, str]] = None
 63    user_agent: Optional[str] = None
 64    config: Optional[Dict] = None
 65    cookie_names: Optional[List[str]] = None
 66    _client: httpx.Client = field(init=False, repr=False)
 67
 68    def __post_init__(self) -> None:
 69        """Initialize VintedWrapper after dataclass initialization.
 70
 71        Validates the base URL, sets up user agent, initializes httpx client,
 72        and fetches session cookies if not provided.
 73
 74        Raises:
 75            RuntimeError: If the base URL is invalid.
 76        """
 77        if not url_validator(self.baseurl):
 78            _log.error("'%s' is not a valid url", self.baseurl)
 79            raise RuntimeError(f"'{self.baseurl}' is not a valid url, please check it!")
 80
 81        log_constructor(
 82            log=_log,
 83            self=self,
 84            baseurl=self.baseurl,
 85            user_agent=self.user_agent,
 86            session_cookie=self.session_cookie,
 87            config=self.config,
 88        )
 89
 90        if self.user_agent is None:
 91            self.user_agent = get_random_user_agent()
 92        if self.cookie_names is None:
 93            self.cookie_names = [SESSION_COOKIE_NAME]
 94        self._client = httpx.Client(**get_httpx_config(self.baseurl, self.config))
 95        if self.session_cookie is None:
 96            self.session_cookie = self.refresh_cookie()
 97
 98    def refresh_cookie(self, retries: int = DEFAULT_RETRIES) -> Dict[str, str]:
 99        """Manually refresh the session cookie.
100
101        Args:
102            retries: Number of retry attempts (default: 3).
103
104        Returns:
105            Dictionary containing session cookies.
106
107        Raises:
108            RuntimeError: If cookies cannot be fetched after all retries.
109        """
110        log_refresh_cookie(_log)
111        return VintedWrapper.fetch_cookie(
112            self._client,
113            get_cookie_headers(self.baseurl, self.user_agent),
114            self.cookie_names,
115            retries,
116        )
117
118    @staticmethod
119    def fetch_cookie(
120        client: httpx.Client,
121        headers: Dict,
122        cookie_names: List[str],
123        retries: int = DEFAULT_RETRIES,
124    ) -> Dict[str, str]:
125        """Fetch session cookies from Vinted using HTTP GET request.
126
127        Args:
128            client: httpx.Client instance.
129            headers: HTTP headers dictionary.
130            cookie_names: List of cookie names to extract.
131            retries: Number of retry attempts (default: 3).
132
133        Returns:
134            Dictionary of extracted session cookies.
135
136        Raises:
137            RuntimeError: If cookies cannot be fetched after all retries.
138        """
139        response = None
140
141        for i in range(retries):
142            log_interaction(_log, i, retries)
143            response = client.get("/", headers=headers)
144
145            if response.status_code == HTTP_OK:
146                cookies = extract_cookie_from_response(response, cookie_names)
147                if cookies:
148                    log_cookie_fetched(_log, str(cookies))
149                    return cookies
150                _log.warning("Cannot find session cookie in response")
151            else:
152                log_cookie_fetch_failed(_log, response.status_code, i, retries)
153                sleep_time = RETRY_BASE_SLEEP**i
154                log_sleep(_log, sleep_time)
155                time.sleep(sleep_time)
156
157        _log.error("Cannot fetch session cookie from %s", client.base_url)
158        raise RuntimeError(
159            f"Cannot fetch session cookie from {client.base_url}, because of "
160            f"status code: {response.status_code if response is not None else 'none'}"
161            "different from 200."
162        )
163
164    def search(self, params: Optional[Dict] = None) -> Dict[str, Any]:
165        """Search for items on Vinted.
166
167        Args:
168            params: Query parameters. Common parameters:
169                - search_text (str): Search query
170                - page (int): Page number
171                - per_page (int): Items per page
172                - price_from (float): Minimum price
173                - price_to (float): Maximum price
174                - order (str): Sort order
175                - catalog_ids (str): Category IDs
176                - brand_ids (str): Brand IDs
177                - size_ids (str): Size IDs
178
179        Returns:
180            Dictionary containing JSON response with search results.
181        """
182        log_search(_log, params)
183        return self.curl(API_CATALOG_ITEMS, params=params)
184
185    def item(self, item_id: str, params: Optional[Dict] = None) -> Dict[str, Any]:
186        """Retrieve detailed information about a specific item.
187
188        Args:
189            item_id: The unique identifier of the item.
190            params: Optional query parameters.
191
192        Returns:
193            Dictionary containing JSON response with item details.
194
195        Raises:
196            RuntimeError: If the item is not found or API returns an error.
197
198        Note:
199            It returns a 403 error after a few uses.
200            See: https://github.com/Giglium/vinted_scraper/issues/59
201        """
202        log_item(_log, item_id, params)
203        return self.curl(f"{API_ITEMS}/{item_id}/details", params=params)
204
205    def curl(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]:
206        """Send a custom HTTP GET request to any Vinted API endpoint.
207
208        Automatically handles headers, cookies, retries, and error responses.
209
210        Args:
211            endpoint: API endpoint path (e.g., "/api/v2/users/username").
212            params: Optional query parameters.
213
214        Returns:
215            Dictionary containing the parsed JSON response.
216
217        Raises:
218            RuntimeError: If response status is not 200 or JSON parsing fails.
219        """
220        headers = get_curl_headers(self.baseurl, self.user_agent, self.session_cookie)
221
222        # Logging request
223        log_curl_request(_log, self.baseurl, endpoint, headers, params)
224
225        response = self._client.get(
226            endpoint,
227            headers=get_curl_headers(
228                self.baseurl, self.user_agent, self.session_cookie
229            ),
230            params=params,
231        )
232
233        # Logging response
234        log_curl_response(
235            _log, endpoint, response.status_code, response.headers, response.text
236        )
237
238        # Success
239        if response.status_code == HTTP_OK:
240            try:
241                return response.json()
242            except ValueError as e:
243                _log.error("Failed to parse JSON response from %s: %s", endpoint, e)
244                raise RuntimeError(f"Invalid JSON response from {endpoint}: {e}") from e
245
246        # Fetch (maybe is expired?) the session cookie again and retry the API call
247        if response.status_code == HTTP_UNAUTHORIZED:
248            log_cookie_retry(_log, response.status_code)
249            self.session_cookie = self.refresh_cookie()
250            return self.curl(endpoint, params)
251        raise RuntimeError(
252            f"Cannot perform API call to endpoint {endpoint}, error code: {response.status_code}"
253        )
254
255    def __enter__(self) -> "VintedWrapper":
256        """Enter context manager.
257
258        Returns:
259            Self for use in with statement.
260        """
261        return self
262
263    def __exit__(self, exc_type, exc_val, exc_tb) -> None:  # pragma: no cover
264        """Exit context manager and close HTTP client.
265
266        Args:
267            exc_type: Exception type (unused).
268            exc_val: Exception value (unused).
269            exc_tb: Exception traceback (unused).
270        """
271        self._client.close()

16@dataclass
17class AsyncVintedScraper(AsyncVintedWrapper):
18    """Asynchronous Vinted scraper with typed model support.
19
20    Returns structured VintedItem objects instead of raw JSON dictionaries.
21    Inherits all functionality from AsyncVintedWrapper.
22
23    Example:
24        See https://github.com/Giglium/vinted_scraper/blob/main/examples/async_scraper.py
25    """
26
27    async def search(self, params: Optional[Dict] = None) -> List[VintedItem]:
28        """Search for items on Vinted asynchronously.
29
30        Args:
31            params: Query parameters for the search. Common parameters:
32                - search_text: Search query
33                - page: Page number
34                - per_page: Items per page
35                - price_from: Minimum price
36                - price_to: Maximum price
37                - order: Sort order
38                - catalog_ids: Category IDs
39                - brand_ids: Brand IDs
40                - size_ids: Size IDs
41
42        Returns:
43            List of VintedItem objects representing search results.
44        """
45        response = await super().search(params)
46        return [VintedItem(json_data=item) for item in response["items"]]
47
48    async def item(self, item_id: str, params: Optional[Dict] = None) -> VintedItem:
49        """Retrieve detailed information about a specific item asynchronously.
50
51        Args:
52            item_id: The unique identifier of the item.
53            params: Optional query parameters.
54
55        Returns:
56            VintedItem object with detailed item information including seller details.
57
58        Raises:
59            RuntimeError: If the item is not found or API returns an error.
60
61        Note:
62             It returns a 403 error after a few uses.
63             See: https://github.com/Giglium/vinted_scraper/issues/59
64        """
65        response = await super().item(item_id, params)
66        return VintedItem(json_data=response["item"])
67
68    async def curl(
69        self, endpoint: str, params: Optional[Dict] = None
70    ) -> VintedJsonModel:
71        """Send an async HTTP GET request to any Vinted API endpoint.
72
73        Args:
74            endpoint: The API endpoint path (e.g., "/api/v2/users/username").
75            params: Optional query parameters.
76
77        Returns:
78            VintedJsonModel containing the JSON response.
79
80        Raises:
81            RuntimeError: If the request fails or returns a non-200 status code.
82
83        """
84        response = await super().curl(endpoint, params)
85        return VintedJsonModel(json_data=response)

16@dataclass
17class VintedScraper(VintedWrapper):
18    """Synchronous Vinted scraper with typed model support.
19
20    Returns structured VintedItem objects instead of raw JSON dictionaries.
21    Inherits all functionality from VintedWrapper.
22
23    Example:
24       See https://github.com/Giglium/vinted_scraper/blob/main/examples/scraper.py
25    """
26
27    def search(self, params: Optional[Dict] = None) -> List[VintedItem]:  # type: ignore
28        """Search for items on Vinted.
29
30        Args:
31            params: Query parameters for the search. Common parameters:
32                - search_text: Search query
33                - page: Page number
34                - per_page: Items per page
35                - price_from: Minimum price
36                - price_to: Maximum price
37                - order: Sort order
38                - catalog_ids: Category IDs
39                - brand_ids: Brand IDs
40                - size_ids : Size IDs
41
42        Returns:
43            List of VintedItem objects representing search results.
44        """
45        return [VintedItem(json_data=item) for item in super().search(params)["items"]]
46
47    def item(self, item_id: str, params: Optional[Dict] = None) -> VintedItem:  # type: ignore
48        """Retrieve detailed information about a specific item.
49
50        Args:
51            item_id: The unique identifier of the item.
52            params: Optional query parameters.
53
54        Returns:
55            VintedItem object with detailed item information including seller details.
56
57        Raises:
58            RuntimeError: If the item is not found or API returns an error.
59
60        Note:
61            It returns a 403 error after a few uses.
62            See: https://github.com/Giglium/vinted_scraper/issues/59
63        """
64        return VintedItem(json_data=super().item(item_id, params)["item"])
65
66    def curl(self, endpoint: str, params: Optional[Dict] = None) -> VintedJsonModel:  # type: ignore
67        """Send a custom HTTP GET request to any Vinted API endpoint.
68
69        Args:
70            endpoint: The API endpoint path (e.g., "/api/v2/users/username").
71            params: Optional query parameters.
72
73        Returns:
74            VintedJsonModel containing the JSON response.
75
76        Raises:
77            RuntimeError: If the request fails or returns a non-200 status code.
78        """
79        response = super().curl(endpoint, params)
80        return VintedJsonModel(json_data=response)