Network

HTTP client and session utilities.

Basic Usage

from naminter import CurlCFFISession

async with CurlCFFISession(
    timeout=30,
    impersonate="chrome",
    proxies="http://proxy:8080"
) as http_client:
    response = await http_client.get("https://example.com")

Custom Session Implementation

Implement BaseSession (see API Reference) to use another HTTP client (e.g. aiohttp, httpx). Raise HttpSessionError or HttpError as appropriate.

Example: aiohttp

import asyncio
import time
import aiohttp
from collections.abc import Mapping
from datetime import timedelta
from naminter import (
    BaseSession,
    HttpError,
    HttpSessionError,
    Naminter,
    WMNResponse,
)

class AiohttpSession:
    """Custom aiohttp-based session implementation."""

    def __init__(self, timeout: int = 30, **kwargs):
        self._timeout = aiohttp.ClientTimeout(total=timeout)
        self._session: aiohttp.ClientSession | None = None
        self._kwargs = kwargs

    async def open(self) -> None:
        """Open the aiohttp session."""
        if self._session is None:
            try:
                self._session = aiohttp.ClientSession(
                    timeout=self._timeout,
                    **self._kwargs
                )
            except Exception as e:
                raise HttpSessionError(f"Failed to create session: {e}") from e

    async def close(self) -> None:
        """Close the aiohttp session."""
        if self._session:
            await self._session.close()
            self._session = None

    async def get(
        self, url: str, headers: Mapping[str, str] | None = None
    ) -> WMNResponse:
        """Perform HTTP GET request."""
        return await self.request("GET", url, headers=headers)

    async def post(
        self,
        url: str,
        headers: Mapping[str, str] | None = None,
        data: str | bytes | None = None,
    ) -> WMNResponse:
        """Perform HTTP POST request."""
        return await self.request("POST", url, headers=headers, data=data)

    async def request(
        self,
        method: str,
        url: str,
        headers: Mapping[str, str] | None = None,
        data: str | bytes | None = None,
    ) -> WMNResponse:
        """Perform generic HTTP request."""
        await self.open()

        if self._session is None:
            raise HttpSessionError("Session not initialized")

        try:
            start = time.monotonic()
            async with self._session.request(
                method=method,
                url=url,
                headers=dict(headers) if headers else None,
                data=data,
            ) as response:
                text = await response.text()
                elapsed = timedelta(seconds=time.monotonic() - start)
                return WMNResponse(
                    status_code=response.status,
                    text=text,
                    elapsed=elapsed,
                )
        except asyncio.TimeoutError as e:
            raise HttpError(f"{method} timeout for {url}") from e
        except aiohttp.ClientError as e:
            raise HttpError(f"{method} failed for {url}: {e}") from e
        except Exception as e:
            raise HttpError(f"Unexpected error: {e}") from e

    async def __aenter__(self) -> "AiohttpSession":
        """Async context manager entry."""
        await self.open()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Async context manager exit."""
        await self.close()

# Usage with Naminter
async with AiohttpSession() as http_client:
    async with Naminter(http_client=http_client, wmn_data=wmn_data) as naminter:
        async for result in naminter.enumerate_usernames(["username"]):
            print(f"{result.name}: {result.status.value}")

API Reference

naminter.core.network.BaseSession

Bases: Protocol

Async HTTP client protocol for Naminter adapters.

Implementations should raise the following exceptions: - HttpSessionError: For session initialization/management errors - HttpError: For other network-related errors

All exceptions should preserve the underlying cause when available.

open `async`

open()

Open the underlying HTTP session.

RAISES	DESCRIPTION
`HttpSessionError`	If session initialization fails.

close `async`

close()

Close the underlying HTTP session.

Should handle errors gracefully and not raise exceptions during cleanup, except for CancelledError which must be propagated.

RAISES	DESCRIPTION
`CancelledError`	Propagated to allow proper cancellation handling.

get `async`

get(url, headers=None)

HTTP GET request.

PARAMETER	DESCRIPTION
`url`	The URL to request. TYPE: `str`
`headers`	Optional HTTP headers to include. TYPE: `Mapping[str, str] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`WMNResponse`	Response with status, text, and elapsed time. TYPE: `WMNResponse`

RAISES	DESCRIPTION
`HttpSessionError`	If session is not initialized or invalid.
`HttpError`	For other network-related errors.

post `async`

post(url, headers=None, data=None)

HTTP POST request.

PARAMETER	DESCRIPTION
`url`	The URL to request. TYPE: `str`
`headers`	Optional HTTP headers to include. TYPE: `Mapping[str, str] \| None` DEFAULT: `None`
`data`	Optional request body data. TYPE: `str \| bytes \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`WMNResponse`	Response with status, text, and elapsed time. TYPE: `WMNResponse`

RAISES	DESCRIPTION
`HttpSessionError`	If session is not initialized or invalid.
`HttpError`	For other network-related errors.

request `async`

request(method, url, headers=None, data=None)

Generic HTTP request.

PARAMETER	DESCRIPTION
`method`	HTTP method (GET or POST only). TYPE: `HttpMethod`
`url`	The URL to request. TYPE: `str`
`headers`	Optional HTTP headers to include. TYPE: `Mapping[str, str] \| None` DEFAULT: `None`
`data`	Optional request body data. TYPE: `str \| bytes \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`WMNResponse`	Response with status, text, and elapsed time. TYPE: `WMNResponse`

RAISES	DESCRIPTION
`HttpSessionError`	If session is not initialized or invalid.
`HttpError`	For other network-related errors.

naminter.core.network.CurlCFFISession

CurlCFFISession(*, proxies=None, verify=True, timeout=HTTP_TIMEOUT, allow_redirects=True, impersonate=None, ja3=None, akamai=None, extra_fp=None)

HTTP session implementation using curl_cffi library.

Provides browser impersonation, proxy support, SSL verification, and custom fingerprinting capabilities.

Initialize CurlCFFISession with configuration.

PARAMETER	DESCRIPTION
`proxies`	Proxy configuration as string or dict. TYPE: `str \| dict[str, str] \| None` DEFAULT: `None`
`verify`	Whether to verify SSL certificates. TYPE: `bool` DEFAULT: `True`
`timeout`	Request timeout in seconds. TYPE: `int` DEFAULT: `HTTP_TIMEOUT`
`allow_redirects`	Whether to follow HTTP redirects. TYPE: `bool` DEFAULT: `True`
`impersonate`	Browser to impersonate (e.g., 'chrome', 'firefox'). TYPE: `BrowserTypeLiteral \| None` DEFAULT: `None`
`ja3`	JA3 fingerprint string for TLS fingerprinting. TYPE: `str \| None` DEFAULT: `None`
`akamai`	Akamai fingerprint string. TYPE: `str \| None` DEFAULT: `None`
`extra_fp`	Additional fingerprinting options. TYPE: `ExtraFingerprints \| dict[str, Any] \| None` DEFAULT: `None`

open `async`

open()

Open the HTTP session.

RAISES	DESCRIPTION
`HttpSessionError`	If session initialization fails.

close `async`

close()

Close the HTTP session.

Handles errors gracefully during cleanup. Catches session closure errors and logs them without propagating.

RAISES	DESCRIPTION
`CancelledError`	Re-raised to allow proper cancellation handling.

get `async`

get(url, headers=None)

HTTP GET request.

PARAMETER	DESCRIPTION
`url`	The URL to request. TYPE: `str`
`headers`	Optional HTTP headers to include. TYPE: `Mapping[str, str] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`WMNResponse`	Response with status, text, and elapsed time. TYPE: `WMNResponse`

RAISES	DESCRIPTION
`HttpSessionError`	If session is not initialized or was closed.
`HttpError`	For other network-related errors.

post `async`

post(url, headers=None, data=None)

HTTP POST request.

PARAMETER	DESCRIPTION
`url`	The URL to request. TYPE: `str`
`headers`	Optional HTTP headers to include. TYPE: `Mapping[str, str] \| None` DEFAULT: `None`
`data`	Optional request body data. TYPE: `str \| bytes \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`WMNResponse`	Response with status, text, and elapsed time. TYPE: `WMNResponse`

RAISES	DESCRIPTION
`HttpSessionError`	If session is not initialized or was closed.
`HttpError`	For other network-related errors.

request `async`

request(method, url, headers=None, data=None)

Perform HTTP request.

PARAMETER	DESCRIPTION
`method`	HTTP method (GET or POST only). TYPE: `HttpMethod`
`url`	The URL to request. TYPE: `str`
`headers`	Optional HTTP headers. TYPE: `Mapping[str, str] \| None` DEFAULT: `None`
`data`	Optional request body. TYPE: `str \| bytes \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`WMNResponse`	Response with status, text, and elapsed time. TYPE: `WMNResponse`

RAISES	DESCRIPTION
`HttpSessionError`	If session is not initialized or was closed.
`HttpError`	For unsupported HTTP methods or other network-related errors.

Network

Basic Usage

Custom Session Implementation

Example: aiohttp

API Reference

naminter.core.network.BaseSession

open async

close async

get async

post async

request async

naminter.core.network.CurlCFFISession

open async

close async

get async

post async

request async

open `async`

close `async`

get `async`

post `async`

request `async`

open `async`

close `async`

get `async`

post `async`

request `async`