vinted_scraper

 19@dataclass
 20class AsyncVintedWrapper(BaseVintedWrapper):
 21    """Asynchronous Vinted API wrapper returning raw JSON responses.
 22
 23    Handles cookie management, retries, and async HTTP requests automatically.
 24    Returns raw JSON dictionaries instead of typed objects.
 25
 26    Attributes:
 27        baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 28        session_cookie: Session cookie dict. Auto-fetched if None.
 29        user_agent: Custom user agent string. Auto-generated if None.
 30        config: httpx client configuration dict.
 31        cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 32
 33    Example:
 34        See https://github.com/Giglium/vinted_scraper/blob/main/examples/async_wrapper.py
 35    """
 36
 37    _client: httpx.AsyncClient = field(init=False, repr=False)
 38
 39    @classmethod
 40    async def create(
 41        cls,
 42        baseurl: str,
 43        user_agent: Optional[str] = None,
 44        config: Optional[Dict] = None,
 45        cookie_names: Optional[List[str]] = None,
 46    ):
 47        """Factory method to create an AsyncVintedWrapper instance.
 48
 49        Use this instead of direct instantiation to automatically fetch the session cookie.
 50
 51        Args:
 52            baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 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        Returns:
 58            Initialized AsyncVintedWrapper instance with fetched cookies.
 59        """
 60        _log.debug("Creating the async wrapper using the factory method")
 61        self = cls(
 62            baseurl, user_agent=user_agent, config=config, cookie_names=cookie_names
 63        )
 64        self.session_cookie = await self.refresh_cookie()
 65        return self
 66
 67    def __post_init__(self) -> None:
 68        """Initialize AsyncVintedWrapper after dataclass initialization.
 69
 70        Validates the base URL, sets up user agent, and initializes httpx async client.
 71
 72        Raises:
 73            RuntimeError: If the base URL is invalid.
 74
 75        Note:
 76            Use the create() factory method instead of direct instantiation to
 77            automatically fetch the session cookie.
 78        """
 79        httpx_config = self._validate_and_init()
 80        self._client = httpx.AsyncClient(**httpx_config)
 81
 82    async def refresh_cookie(self, retries: int = DEFAULT_RETRIES) -> Dict[str, str]:
 83        """Manually refresh the session cookie asynchronously.
 84
 85        Args:
 86            retries: Number of retry attempts (default: 3).
 87
 88        Returns:
 89            Dictionary containing session cookies.
 90
 91        Raises:
 92            RuntimeError: If cookies cannot be fetched after all retries.
 93        """
 94        self._log_refresh_cookie()
 95        return await AsyncVintedWrapper.fetch_cookie(
 96            self._client,
 97            self._get_cookie_headers(),
 98            self.cookie_names,
 99            retries,
100        )
101
102    @staticmethod
103    async def fetch_cookie(
104        client: httpx.AsyncClient,
105        headers: Dict,
106        cookie_names: List[str],
107        retries: int = DEFAULT_RETRIES,
108    ) -> Dict[str, str]:
109        """Fetch session cookies from Vinted using async HTTP GET request.
110
111        Args:
112            client: httpx.AsyncClient instance.
113            headers: HTTP headers dictionary.
114            cookie_names: List of cookie names to extract.
115            retries: Number of retry attempts (default: 3).
116
117        Returns:
118            Dictionary of extracted session cookies.
119
120        Raises:
121            RuntimeError: If cookies cannot be fetched after all retries.
122        """
123        response = None
124
125        for i in range(retries):
126            BaseVintedWrapper._log_cookie_interaction(i, retries)
127            response = await client.get("/", headers=headers)
128
129            cookies = BaseVintedWrapper._process_cookie_response(response, cookie_names)
130            if cookies:
131                return cookies
132
133            if response.status_code != HTTP_OK:
134                sleep_time = BaseVintedWrapper._handle_cookie_failure(
135                    response, i, retries
136                )
137                await asyncio.sleep(sleep_time)
138
139        BaseVintedWrapper._raise_cookie_error(client.base_url, response)
140
141    async def search(self, params: Optional[Dict] = None) -> Dict[str, Any]:
142        """Search for items on Vinted asynchronously.
143
144        Args:
145            params: Query parameters. Common parameters:
146                - search_text: Search query
147                - page: Page number
148                - per_page: Items per page
149                - price_from: Minimum price
150                - price_to: Maximum price
151                - order: Sort order
152                - catalog_ids: Category IDs
153                - brand_ids: Brand IDs
154                - size_ids: Size IDs
155
156        Returns:
157            Dictionary containing JSON response with search results.
158        """
159        self._log_search(params)
160        return await self.curl(self._search_endpoint(), params=params)
161
162    async def item(self, item_id: str, params: Optional[Dict] = None) -> Dict[str, Any]:
163        """Retrieve detailed information about a specific item asynchronously.
164
165        Args:
166            item_id: The unique identifier of the item.
167            params: Optional query parameters.
168
169        Returns:
170            Dictionary containing JSON response with item details.
171
172        Raises:
173            RuntimeError: If the item is not found or API returns an error.
174
175        Note:
176            It returns a 403 error after a few uses.
177            See: https://github.com/Giglium/vinted_scraper/issues/59
178        """
179        self._log_item(item_id, params)
180        return await self.curl(self._item_endpoint(item_id), params=params)
181
182    async def curl(
183        self,
184        endpoint: str,
185        params: Optional[Dict] = None,
186        *,
187        _retries: int = 0,
188    ) -> Dict[str, Any]:
189        """Send an async HTTP GET request to any Vinted API endpoint.
190
191        Automatically handles headers, cookies, retries, and error responses.
192
193        Args:
194            endpoint: API endpoint path (e.g., "/api/v2/users/username").
195            params: Optional query parameters.
196
197        Returns:
198            Dictionary containing the parsed JSON response.
199
200        Raises:
201            RuntimeError: If response status is not 200 or JSON parsing fails.
202        """
203        headers = self._build_curl_headers()
204        self._log_curl_request(endpoint, headers, params)
205
206        response = await self._client.get(endpoint, headers=headers, params=params)
207
208        self._log_curl_response(
209            endpoint, response.status_code, response.headers, response.text
210        )
211
212        if response.status_code == HTTP_OK:
213            return self._handle_curl_response(response, endpoint)
214
215        if response.status_code == HTTP_UNAUTHORIZED and _retries < DEFAULT_RETRIES:
216            self._log_cookie_retry(response.status_code)
217            self.session_cookie = await self.refresh_cookie()
218            return await self.curl(endpoint, params, _retries=_retries + 1)
219
220        self._raise_curl_error(endpoint, response.status_code)
221
222    async def __aenter__(self) -> "AsyncVintedWrapper":  # pragma: no cover
223        """Enter async context manager.
224
225        Returns:
226            Self for use in async with statement.
227        """
228        return self
229
230    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:  # pragma: no cover
231        """Exit async context manager and close HTTP client.
232
233        Args:
234            exc_type: Exception type (unused).
235            exc_val: Exception value (unused).
236            exc_tb: Exception traceback (unused).
237        """
238        await self._client.aclose()

 19@dataclass
 20class VintedWrapper(BaseVintedWrapper):
 21    """Synchronous Vinted API wrapper returning raw JSON responses.
 22
 23    Handles cookie management, retries, and HTTP requests automatically.
 24    Returns raw JSON dictionaries instead of typed objects.
 25
 26    Attributes:
 27        baseurl: Vinted domain URL (e.g., "https://www.vinted.com").
 28        session_cookie: Session cookie dict. Auto-fetched if None.
 29        user_agent: Custom user agent string. Auto-generated if None.
 30        config: httpx client configuration dict.
 31        cookie_names: List of cookie names to extract. Defaults to ["access_token_web"].
 32
 33    Example:
 34        See https://github.com/Giglium/vinted_scraper/blob/main/examples/wrapper.py
 35    """
 36
 37    _client: httpx.Client = field(init=False, repr=False)
 38
 39    def __post_init__(self) -> None:
 40        """Initialize VintedWrapper after dataclass initialization.
 41
 42        Validates the base URL, sets up user agent, initializes httpx client,
 43        and fetches session cookies if not provided.
 44
 45        Raises:
 46            RuntimeError: If the base URL is invalid.
 47        """
 48        httpx_config = self._validate_and_init()
 49        self._client = httpx.Client(**httpx_config)
 50        if self.session_cookie is None:
 51            self.session_cookie = self.refresh_cookie()
 52
 53    def refresh_cookie(self, retries: int = DEFAULT_RETRIES) -> Dict[str, str]:
 54        """Manually refresh the session cookie.
 55
 56        Args:
 57            retries: Number of retry attempts (default: 3).
 58
 59        Returns:
 60            Dictionary containing session cookies.
 61
 62        Raises:
 63            RuntimeError: If cookies cannot be fetched after all retries.
 64        """
 65        self._log_refresh_cookie()
 66        return VintedWrapper.fetch_cookie(
 67            self._client,
 68            self._get_cookie_headers(),
 69            self.cookie_names,
 70            retries,
 71        )
 72
 73    @staticmethod
 74    def fetch_cookie(
 75        client: httpx.Client,
 76        headers: Dict,
 77        cookie_names: List[str],
 78        retries: int = DEFAULT_RETRIES,
 79    ) -> Dict[str, str]:
 80        """Fetch session cookies from Vinted using HTTP GET request.
 81
 82        Args:
 83            client: httpx.Client instance.
 84            headers: HTTP headers dictionary.
 85            cookie_names: List of cookie names to extract.
 86            retries: Number of retry attempts (default: 3).
 87
 88        Returns:
 89            Dictionary of extracted session cookies.
 90
 91        Raises:
 92            RuntimeError: If cookies cannot be fetched after all retries.
 93        """
 94        response = None
 95
 96        for i in range(retries):
 97            BaseVintedWrapper._log_cookie_interaction(i, retries)
 98            response = client.get("/", headers=headers)
 99
100            cookies = BaseVintedWrapper._process_cookie_response(response, cookie_names)
101            if cookies:
102                return cookies
103
104            if response.status_code != HTTP_OK:
105                sleep_time = BaseVintedWrapper._handle_cookie_failure(
106                    response, i, retries
107                )
108                time.sleep(sleep_time)
109
110        BaseVintedWrapper._raise_cookie_error(client.base_url, response)
111
112    def search(self, params: Optional[Dict] = None) -> Dict[str, Any]:
113        """Search for items on Vinted.
114
115        Args:
116            params: Query parameters. Common parameters:
117                - search_text (str): Search query
118                - page (int): Page number
119                - per_page (int): Items per page
120                - price_from (float): Minimum price
121                - price_to (float): Maximum price
122                - order (str): Sort order
123                - catalog_ids (str): Category IDs
124                - brand_ids (str): Brand IDs
125                - size_ids (str): Size IDs
126
127        Returns:
128            Dictionary containing JSON response with search results.
129        """
130        self._log_search(params)
131        return self.curl(self._search_endpoint(), params=params)
132
133    def item(self, item_id: str, params: Optional[Dict] = None) -> Dict[str, Any]:
134        """Retrieve detailed information about a specific item.
135
136        Args:
137            item_id: The unique identifier of the item.
138            params: Optional query parameters.
139
140        Returns:
141            Dictionary containing JSON response with item details.
142
143        Raises:
144            RuntimeError: If the item is not found or API returns an error.
145
146        Note:
147            It returns a 403 error after a few uses.
148            See: https://github.com/Giglium/vinted_scraper/issues/59
149        """
150        self._log_item(item_id, params)
151        return self.curl(self._item_endpoint(item_id), params=params)
152
153    def curl(
154        self,
155        endpoint: str,
156        params: Optional[Dict] = None,
157        *,
158        _retries: int = 0,
159    ) -> Dict[str, Any]:
160        """Send a custom HTTP GET request to any Vinted API endpoint.
161
162        Automatically handles headers, cookies, retries, and error responses.
163
164        Args:
165            endpoint: API endpoint path (e.g., "/api/v2/users/username").
166            params: Optional query parameters.
167
168        Returns:
169            Dictionary containing the parsed JSON response.
170
171        Raises:
172            RuntimeError: If response status is not 200 or JSON parsing fails.
173        """
174        headers = self._build_curl_headers()
175        self._log_curl_request(endpoint, headers, params)
176
177        response = self._client.get(endpoint, headers=headers, params=params)
178
179        self._log_curl_response(
180            endpoint, response.status_code, response.headers, response.text
181        )
182
183        if response.status_code == HTTP_OK:
184            return self._handle_curl_response(response, endpoint)
185
186        if response.status_code == HTTP_UNAUTHORIZED and _retries < DEFAULT_RETRIES:
187            self._log_cookie_retry(response.status_code)
188            self.session_cookie = self.refresh_cookie()
189            return self.curl(endpoint, params, _retries=_retries + 1)
190
191        self._raise_curl_error(endpoint, response.status_code)
192
193    def __enter__(self) -> "VintedWrapper":
194        """Enter context manager.
195
196        Returns:
197            Self for use in with statement.
198        """
199        return self
200
201    def __exit__(self, exc_type, exc_val, exc_tb) -> None:  # pragma: no cover
202        """Exit context manager and close HTTP client.
203
204        Args:
205            exc_type: Exception type (unused).
206            exc_val: Exception value (unused).
207            exc_tb: Exception traceback (unused).
208        """
209        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(self, item_id: str, params: Optional[Dict] = None) -> VintedItem:
47        """Retrieve detailed information about a specific item asynchronously.
48
49        Args:
50            item_id: The unique identifier of the item.
51            params: Optional query parameters.
52
53        Returns:
54            VintedItem object with detailed item information including seller details.
55
56        Raises:
57            RuntimeError: If the item is not found or API returns an error.
58
59        Note:
60             It returns a 403 error after a few uses.
61             See: https://github.com/Giglium/vinted_scraper/issues/59
62        """
63        response = await super().item(item_id, params)
64        return VintedItem(json_data=response["item"])
65
66    async def curl(
67        self, endpoint: str, params: Optional[Dict] = None, *, _retries: int = 0
68    ) -> VintedJsonModel:
69        """Send an async HTTP GET request to any Vinted API endpoint.
70
71        Args:
72            endpoint: The API endpoint path (e.g., "/api/v2/users/username").
73            params: Optional query parameters.
74
75        Returns:
76            VintedJsonModel containing the JSON response.
77
78        Raises:
79            RuntimeError: If the request fails or returns a non-200 status code.
80        """
81        response = await super().curl(endpoint, params, _retries=_retries)
82        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(self, item_id: str, params: Optional[Dict] = None) -> VintedItem:  # type: ignore
46        """Retrieve detailed information about a specific item.
47
48        Args:
49            item_id: The unique identifier of the item.
50            params: Optional query parameters.
51
52        Returns:
53            VintedItem object with detailed item information including seller details.
54
55        Raises:
56            RuntimeError: If the item is not found or API returns an error.
57
58        Note:
59            It returns a 403 error after a few uses.
60            See: https://github.com/Giglium/vinted_scraper/issues/59
61        """
62        return VintedItem(json_data=super().item(item_id, params)["item"])
63
64    def curl(
65        self, endpoint: str, params: Optional[Dict] = None, *, _retries: int = 0
66    ) -> 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, _retries=_retries)
80        return VintedJsonModel(json_data=response)