from __future__ import annotations
from typing import Any
import tls_client
from ._http import HttpClient, HttpClientConfig
from .resources import (
CreditsResource,
FeedResource,
MembersResource,
MessagesResource,
NotificationsResource,
ReputationResource,
SearchResource,
SessionResource,
ThreadsResource,
UserCPResource,
UsersResource,
)
__all__ = ['OGUClient']
[docs]
class OGUClient:
"""The top-level OGUsers client.
Aggregates 11 resource groups (``session``, ``users``, ``usercp``,
``reputation``, ``credits``, ``messages``, ``notifications``, ``feed``,
``threads``, ``search``, ``members``) over a single ``tls_client``
session with Cloudflare-friendly TLS fingerprints.
Use as an async context manager — the underlying session is closed on
``__aexit__``:
>>> async with OGUClient() as client:
... await client.session.login('username', 'password')
... user = await client.users.get_by_username('forgivenforget')
Args:
proxy: Optional proxy string. Accepts ``user:pass@host:port``,
``host:port:user:pass``, or plain ``host:port``; auto-normalized
to ``http://...``.
base_url: Override the forum base URL (default ``https://oguser.com``).
timeout_seconds: Per-request timeout (default ``30``).
max_retries: How many times to retry on ``429`` and ``5xx`` (default ``0``).
retry_backoff_seconds: Base delay for exponential backoff (default ``0.5``).
client_identifier: TLS fingerprint identifier for ``tls_client``
(default ``'chrome131'``).
config: Provide a fully-built :class:`HttpClientConfig` instead of the
individual kwargs above.
session: Provide a pre-configured ``tls_client.Session``. The client
will not close it automatically.
Attributes:
session (SessionResource): Login / logout / login form.
users (UsersResource): Profile lookup.
usercp (UserCPResource): Account settings.
reputation (ReputationResource): Send reputation.
credits (CreditsResource): Send / track credits.
messages (MessagesResource): Private messages.
notifications (NotificationsResource): Notifications + alerts.
feed (FeedResource): Explore + home feeds.
threads (ThreadsResource): View threads / forums; reply; create.
search (SearchResource): Forum-wide search.
members (MembersResource): Member directory.
"""
def __init__(
self,
*,
proxy: str | None = None,
base_url: str | None = None,
timeout_seconds: float | None = None,
max_retries: int | None = None,
retry_backoff_seconds: float | None = None,
client_identifier: str | None = None,
config: HttpClientConfig | None = None,
session: tls_client.Session | None = None,
) -> None:
resolved_config = self._resolve_config(
config = config,
base_url = base_url,
timeout_seconds = timeout_seconds,
max_retries = max_retries,
retry_backoff_seconds = retry_backoff_seconds,
client_identifier = client_identifier,
)
self._http: HttpClient = HttpClient(
proxy = proxy,
config = resolved_config,
session = session,
)
self.session: SessionResource = SessionResource(self._http)
self.users: UsersResource = UsersResource(self._http)
self.usercp: UserCPResource = UserCPResource(self._http)
self.reputation: ReputationResource = ReputationResource(self._http)
self.credits: CreditsResource = CreditsResource(self._http)
self.messages: MessagesResource = MessagesResource(self._http)
self.notifications: NotificationsResource = NotificationsResource(self._http)
self.feed: FeedResource = FeedResource(self._http)
self.threads: ThreadsResource = ThreadsResource(self._http)
self.search: SearchResource = SearchResource(self._http)
self.members: MembersResource = MembersResource(self._http)
async def __aenter__(self) -> OGUClient:
return self
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
self.close()
[docs]
def close(self) -> None:
"""Close the underlying HTTP session.
Called automatically when used as an async context manager. Only
closes the session if the client created it (i.e. you didn't pass
your own ``session`` to ``__init__``).
"""
self._http.close()
@property
def http(self) -> HttpClient:
"""The underlying :class:`HttpClient`. Use for endpoints not yet wrapped."""
return self._http
@property
def cookies(self) -> Any:
"""The session cookie jar (``tls_client.Session.cookies``).
Iterable of :class:`http.cookiejar.Cookie` objects. Use for
persistence:
>>> jar = {c.name: c.value for c in client.cookies}
"""
return self._http.cookies
@staticmethod
def _resolve_config(
*,
config: HttpClientConfig | None,
base_url: str | None,
timeout_seconds: float | None,
max_retries: int | None,
retry_backoff_seconds: float | None,
client_identifier: str | None,
) -> HttpClientConfig:
if config is None:
base = HttpClientConfig()
else:
base = config
return HttpClientConfig(
base_url = base_url if base_url is not None else base.base_url,
timeout_seconds = (
timeout_seconds if timeout_seconds is not None else base.timeout_seconds
),
max_retries = max_retries if max_retries is not None else base.max_retries,
retry_backoff_seconds = (
retry_backoff_seconds
if retry_backoff_seconds is not None
else base.retry_backoff_seconds
),
client_identifier = (
client_identifier if client_identifier is not None else base.client_identifier
),
default_headers = base.default_headers,
)