Wavyzz/lbc
2
0
Fork 0
mirror of https://github.com/etienne-hd/lbc.git synced 2026-04-16 20:25:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

Changed: proxy handling, you can now remove proxy using client.proxy = None; client._session is now public -> client.session

Browse Source 
Added: proxy examples at examples/proxy.py
This commit is contained in:
etienne-hd
2026-01-05 18:26:22 +01:00
parent bbbdd1652f
commit 6669950f82
8 changed files with 76 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
README.md
View File

@@ -41,7 +41,6 @@ pip install lbc
```
## Usage
**Full documentation will be available soon.**
Start with the [examples](examples/) to quickly understand how to use the library in real-world scenarios.
@@ -56,13 +55,29 @@ client = lbc.Client()
#### Proxy
You can also configure the client to use a proxy by providing a `Proxy` object:
```python
proxy = lbc.Proxy(
host=...,
port=...,
username=...,
password=...
# Setup proxy1
proxy1 = lbc.Proxy(
host="127.0.0.1",
port=12345,
username="username",
password="password",
scheme="http"
)
client = lbc.Client(proxy=proxy)
# Initialize client with proxy1
client = lbc.Client(proxy=proxy1)
# Setup proxy2
proxy2 = lbc.Proxy(
host="127.0.0.1",
port=23456,
)
# Change client proxy to proxy2
client.proxy = proxy2
# Remove proxy
client.proxy = None
```

29
examples/proxy.py Normal file
View File

@@ -0,0 +1,29 @@
import lbc
def main() -> None:
# Setup proxy1
proxy1 = lbc.Proxy(
host="127.0.0.1",
port=12345,
username="username",
password="password",
scheme="http"
)
# Initialize client with proxy1
client = lbc.Client(proxy=proxy1)
# Setup proxy2
proxy2 = lbc.Proxy(
host="127.0.0.1",
port=23456,
)
# Change client proxy to proxy2
client.proxy = proxy2
# Remove proxy
client.proxy = None
if __name__ == "__main__":
main()

9
lbc/client.py
View File

@@ -36,7 +36,7 @@ class Client(
self.timeout = timeout
self.max_retries = max_retries
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, timeout: int = 30, max_retries: int = 5) -> Union[dict, None]:
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, max_retries: int = -1) -> dict:
"""
Internal method to send an HTTP request using the configured session.
@@ -54,19 +54,22 @@ class Client(
Returns:
dict: Parsed JSON response from the server.
"""
if max_retries == -1:
max_retries = self.max_retries
response = self.session.request(
method=method,
url=url,
json=payload,
timeout=timeout,
verify=self.request_verify,
timeout=self.timeout
)
if response.ok:
return response.json()
elif response.status_code == 403:
if max_retries > 0:
self.session = self._init_session(proxy=self._proxy, impersonate=self._impersonate, request_verify=self.request_verify) # Re-init session
return self._fetch(method=method, url=url, payload=payload, timeout=timeout, max_retries=max_retries - 1)
return self._fetch(method=method, url=url, payload=payload, max_retries=max_retries - 1)
if self.proxy:
raise DatadomeError(f"Access blocked by Datadome: your proxy appears to have a poor reputation, try to change it.")
else:

2
lbc/mixin/ad.py
View File

@@ -17,5 +17,5 @@ class AdMixin:
Returns:
Ad: An `Ad` object containing the parsed ad information.
"""
body = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/adfinder/v1/classified/{ad_id}", timeout=self.timeout, max_retries=self.max_retries)
body = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/adfinder/v1/classified/{ad_id}")
return Ad._build(raw=body, client=self)

2
lbc/mixin/search.py
View File

@@ -56,5 +56,5 @@ class SearchMixin:
owner_type=owner_type, shippable=shippable, search_in_title_only=search_in_title_only, **kwargs
)
body = self._fetch(method="POST", url="https://api.leboncoin.fr/finder/search", payload=payload, timeout=self.timeout, max_retries=self.max_retries)
body = self._fetch(method="POST", url="https://api.leboncoin.fr/finder/search", payload=payload)
return Search._build(raw=body, client=self)

34
lbc/mixin/session.py
View File

@@ -6,8 +6,9 @@ import uuid
from ..model import Proxy
class SessionMixin:
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True, **kwargs):
self._session = self._init_session(proxy=proxy, impersonate=impersonate, request_verify=request_verify)
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None,
request_verify: bool = True, **kwargs):
self.session = self._init_session(proxy=proxy, impersonate=impersonate, request_verify=request_verify)
self._proxy = proxy
self._impersonate = impersonate
super().__init__(**kwargs)
@@ -58,7 +59,7 @@ class SessionMixin:
)
session = requests.Session(
impersonate=impersonate,
impersonate=impersonate
)
session.headers.update(
@@ -77,17 +78,6 @@ class SessionMixin:
session.get("https://www.leboncoin.fr/", verify=request_verify) # Init cookies
return session
@property
def session(self) -> requests.Session:
return self._session
@session.setter
def session(self, value: requests.Session):
if isinstance(value, requests.Session):
self._session = value
else:
raise TypeError("Session must be an instance of the curl_cffi.requests.Session")
@property
def proxy(self) -> Proxy:
@@ -95,10 +85,14 @@ class SessionMixin:
@proxy.setter
def proxy(self, value: Proxy):
if isinstance(value, Proxy):
self._session.proxies = {
"http": value.url,
"https": value.url
}
if value:
if isinstance(value, Proxy):
self.session.proxies = {
"http": value.url,
"https": value.url
}
else:
raise TypeError("Proxy must be an instance of the lbc.Proxy")
else:
raise TypeError("Proxy must be an instance of the Proxy class")
self.session.proxies = {}
self._proxy = value

4
lbc/mixin/user.py
View File

@@ -15,12 +15,12 @@ class UserMixin:
Returns:
User: A `User` object containing the parsed user information.
"""
user_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/user-card/v2/{user_id}/infos", timeout=self.timeout, max_retries=self.max_retries)
user_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/user-card/v2/{user_id}/infos")
pro_data = None
if user_data.get("account_type") == "pro":
try:
pro_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/onlinestores/v2/users/{user_id}?fields=all", timeout=self.timeout, max_retries=self.max_retries)
pro_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/onlinestores/v2/users/{user_id}?fields=all")
except NotFoundError:
pass # Some professional users may not have a Leboncoin page.
return User._build(user_data=user_data, pro_data=pro_data)

2
lbc/model/proxy.py
View File

@@ -7,7 +7,7 @@ class Proxy:
port: Union[str, int]
username: Optional[str] = None
password: Optional[str] = None
scheme: Optional[str] = "http"
scheme: str = "http"
@property
def url(self):