eaf_base_api

🏗️ Overview & Version 4

eaf_base_api is the shared engine used by all site-specific API packages. Version 4 separates each responsibility into an explicit, typed interface:

• BaseCore — HTTP sessions, request retries, response decoding, caching, proxy/interface binding, HLS inspection, and downloads
• BaseMedia + media_field — source-aware, atomic lazy loading with deterministic source precedence
• Helper + IteratorConfig — bounded page/item scheduling, ordering, retries, error handlers, and load selection
• ScrapeStream + ScrapeResult — deterministic stream cleanup and immutable success/failure results
• CacheBackend — replaceable storage contract; the built-in cache uses byte limits and TTL expiry

📦 Installation

pip install eaf_base_api

# Install optional HLS parsing/remux dependencies as well
pip install "eaf_base_api[hls]"

Version 4 requires Python 3.12 or newer. The package ships a py.typed marker, so type checkers can consume its inline annotations.

🎛️ RuntimeConfig

Create a RuntimeConfig for each independently configured BaseCore. A process-wide config instance remains available as the default, but a dedicated instance avoids unrelated clients changing one another.

When changes take effect

Cache limits/TTLs and the initial locale header are captured when BaseCore is constructed. Proxy, interface, ip_resolve, TLS, HTTP, DoH, impersonation, cookies, and bandwidth options are captured whenever its session is created. Request budgets, retry delays/multiplier, and request delay are read for each request; HLS timeout/workers are read when a download begins; and omitted IteratorConfig values are resolved whenever a new stream is created.

🔗 Client Integration & Lifecycle

from base_api import BaseCore
from base_api.modules.config import RuntimeConfig
from xvideos_api import Client

runtime = RuntimeConfig()
runtime.timeout = 60
runtime.request_attempts = 5
runtime.request_multiplier = 2.0
runtime.proxy = "socks5://127.0.0.1:9050"
runtime.interface = None  # Or a local interface IP
runtime.cookies = {"session": "value"}

core = BaseCore(configuration=runtime)
client = Client(core=core)

try:
    video = await client.get_video("https://www.xvideos.com/video...")
finally:
    await core.close()

BaseCore starts without an HTTP session. Context-manager entry, the first request, or an explicit initialize_session() creates it lazily; further initialization calls are no-ops while that session remains live. Prefer async with BaseCore(configuration=runtime) as core: for direct use. await core.close() closes and clears the current session, and a later request safely recreates it from the core's current configuration.

🌐 Explicit Request API

Version 4 replaces the flag-driven fetch() method with one method per representation:

from base_api import BaseCore, CachePolicy

async with BaseCore() as core:
    response = await core.request(url)
    text = await core.fetch_text(url)
    data = await core.fetch_bytes(download_url)

    refreshed = await core.fetch_text(
        url,
        cache_policy=CachePolicy.REFRESH,
    )
    uncached = await core.fetch_text(
        url,
        cache_policy=CachePolicy.BYPASS,
    )

CachePolicy.USE reads and writes, REFRESH skips the read and replaces the cached value, and BYPASS neither reads nor writes. All three request methods accept timeout, cookies, redirects, data, HTTP method, headers, JSON, parameters, and retry_non_idempotent.

🔁 BaseCore Request Retries

RuntimeConfig.request_attempts includes the first request. Backoff uses request_retry_initial_delay, request_multiplier as Tenacity's exponential base, request_retry_max_delay, and request_retry_jitter. Idempotent methods (GET, HEAD, PUT, DELETE, OPTIONS, and TRACE) retry network failures plus HTTP 408, 425, 429, and 5xx responses. A 429 response honors Retry-After when available.

• POST/PATCH and other non-idempotent operations are attempted once unless retry_non_idempotent=True is explicitly safe.
• Non-retryable HTTP failures—including 401/403, 404, 410, and 4xx responses other than 408, 425, and 429—are terminal.
• Exhaustion raises RequestRetriesExhausted with .url, .attempts, and .last_error.

🔒 Proxy, Interface & IP Resolution

from base_api.modules.config import RuntimeConfig

runtime = RuntimeConfig()
runtime.proxy = "socks5://127.0.0.1:9050"
runtime.proxy_auth = "username:password"
runtime.interface = "192.0.2.10"
runtime.ip_resolve = 1  # 1 for IPv4, 2 for IPv6, None for dual-stack

# Only disable verification for a proxy you control and trust.
runtime.verify_ssl = False

The old proxies mapping was replaced by the singular proxy URL. interface is passed to curl_cffi.AsyncSession and must be a local interface IP address. ip_resolve allows restricting DNS resolution to IPv4 (1) or IPv6 (2) when dual-stack connectivity issues arise; it also respects the CURL_IPRESOLVE environment variable (e.g. CURL_IPRESOLVE=1 or 4 for IPv4, 2 or 6 for IPv6).

💾 Caching

The built-in Cache uses separate, thread-safe TTL caches for text responses and HLS segment lists. Limits are measured in UTF-8 bytes, not item counts. Request keys include method, URL, redirects, parameters, body, headers, and cookies; sensitive values are fingerprinted rather than stored in plaintext keys.

• Only successful GET text responses are stored.
• Concurrent cache misses for the same request share one in-flight network operation.
• core.cache.clear() clears both built-in caches.
• Pass a custom CacheBackend to BaseCore(cache=...) to replace storage.

🧩 Source-aware BaseMedia

Remote fields are declared with media_field(). The first source has highest precedence. A loader is async, returns a complete mapping, and never mutates the object directly; results are validated and committed atomically.

from dataclasses import dataclass
from typing import ClassVar
from base_api import BaseMedia, media_field

@dataclass(kw_only=True, slots=True)
class Video(BaseMedia):
    title: str | None = media_field("html", "api")
    duration: int | None = media_field("api")

    loader_methods: ClassVar[dict[str, str]] = {
        "html": "_load_html",
        "api": "_load_api",
    }

    async def _load_html(self) -> dict[str, object]:
        data = await fetch_html(self.url)
        return {"title": data.get("title")}

    async def _load_api(self) -> dict[str, object]:
        data = await fetch_api(self.url)
        return {"title": data.get("title"), "duration": data.get("duration")}

Direct access to an unresolved field raises DataNotLoadedError. Use retry_failed=False with a load method when a previously failed source should not be attempted again.

⚙️ IteratorConfig

Site-specific iterator methods now accept one IteratorConfig instead of many concurrency, ordering, loading, and callback parameters. Values left as None are resolved from the active core's RuntimeConfig.

🛡️ RetryPolicy & Custom Error Handling

RetryPolicy.max_attempts includes the first attempt. Without a custom handler, eligible exception classes are selected with retry_for; the delay is bounded exponential backoff plus uniformly random jitter.

from base_api import (
    ErrorAction,
    ErrorMode,
    MediaLoadError,
    MediaLoadErrors,
    ResultOrder,
    RetryPolicy,
    ScrapeErrorContext,
)
from base_api.modules.config import IteratorConfig
from base_api.modules.errors import ResourceGone

def resource_is_gone(error: BaseException) -> bool:
    if isinstance(error, ResourceGone):
        return True
    if isinstance(error, MediaLoadError):
        return resource_is_gone(error.original_error)
    if isinstance(error, MediaLoadErrors):
        return any(resource_is_gone(item) for item in error.errors)
    return False

def handle_page_error(context: ScrapeErrorContext) -> ErrorAction:
    print("page", context.url, context.attempt, context.error)
    return ErrorAction.RETRY

async def handle_item_error(context: ScrapeErrorContext) -> ErrorAction:
    print("item", context.url, context.attempt, context.error)
    if resource_is_gone(context.error):
        return ErrorAction.SKIP
    return ErrorAction.RETRY

iterator_config = IteratorConfig(
    max_page_concurrency=2,
    max_item_concurrency=8,
    order=ResultOrder.ORIGINAL,
    item_retry=RetryPolicy(
        max_attempts=3,
        base_delay=0.5,
        multiplier=2.0,
        max_delay=8.0,
        jitter=0.25,
        retry_for=(Exception,),
    ),
    page_error_handler=handle_page_error,
    item_error_handler=handle_item_error,
    item_error_mode=ErrorMode.YIELD,
    page_error_mode=ErrorMode.SKIP,
    load_specific_sources=("html",),
)

Page and item handlers are independent; each receives only failures from its own stage, and either field may be left unset to use automatic policy for that stage. A sync or async handler runs on every failed stage attempt and may return RETRY, RAISE, YIELD, or SKIP. An explicit RETRY can override retry_for, but it cannot exceed the policy's hard limit; on the final attempt it falls back to the configured error mode. Page failures containing a nested HTTP 404 remain terminal. A handler that raises or returns another value produces a fatal ErrorHandlerError.

🌊 ScrapeStream & ScrapeResult

Helper.iterator() returns a lazily started ScrapeStream async context manager. For convenient direct iteration without manually managing context lifecycles, scrape_stream() provides an AsyncGenerator[ScrapeResult[T], None] that handles internal stream lifecycle automatically.

# Direct iteration via scrape_stream() async generator
from base_api import scrape_stream

async for result in scrape_stream(
    target_page_urls=page_urls,
    item_extractor=extractor,
    iterator_config=iterator_config,
    core=core,
):
    if not result.succeeded:
        print(result.stage, result.url, result.error)
        continue
    media = result.unwrap()  # Returns T or raises result.error

# Alternatively, using Helper.iterator() with an async context manager
stream = helper.iterator(
    target_page_urls=page_urls,
    item_extractor=extractor,
    iterator_config=iterator_config,
)

async with stream:
    async for result in stream:
        if not result.succeeded:
            print(result.stage, result.url, result.error)
            continue
        media = result.unwrap()

⬇️ Download Configurations

Shared fields

DownloadConfigHLS and DownloadConfigRAW share quality, path, callback, no_title, and stop_event. Quality accepts a supported integer height or "best", "half", "worst", and common "720p"-style values. Version 4.1 normalizes landscape and portrait variants by their shorter dimension, chooses the closest available tier for an explicit numeric request (preferring the higher tier on a tie), and uses bandwidth and frame rate as fallbacks when a master playlist omits resolution metadata.

HLS inspection

await core.list_available_qualities(m3u8_url) returns sorted, unique integer tiers from a master playlist. await core.get_m3u8_by_quality(m3u8_url, quality) returns the selected media-playlist URL. Both methods accept a playlist URL, inline master text, or a callable/awaitable resolving to either form. Relative variant paths can be resolved only when the master was fetched from a URL; an inline master must contain absolute variant URLs.

DownloadConfigHLS

BaseCore.download() reads RuntimeConfig.timeout and RuntimeConfig.max_workers_download from that core when dispatching each HLS download. Dedicated cores therefore keep their HLS timeout and worker settings isolated from the process-wide default configuration. HLS cancellation uses an asyncio.Event: setting it cancels pending segment tasks promptly, and progress/remux callbacks are synchronized across worker threads. Version 4.1.1 also normalizes discontinuous packet timestamps while remuxing so valid HLS discontinuities do not abort MP4 output.

DownloadConfigRAW

Direct-file downloads add allow_multipart=True, max_workers=5, read_timeout=120.0, chunk_size=1024, and max_retries=5. These downloader retries are separate from RuntimeConfig.request_attempts.

🤝 Provider Helpers (4.2.0)

base_api.modules.provider centralizes common request and download handling across all 15 site packages, eliminating boilerplate while enforcing uniform exception translation and contextual logging:

from base_api.modules.provider import fetch_content, download_errors, download_hls
from base_api.modules.logger import get_logger
from my_api.modules.errors import DownloadFailed

logger = get_logger(__name__)

async def get_html_content(core, url, *, owner=None):
    return await fetch_content(core, url, logger=logger, owner=owner)

@download_errors(DownloadFailed)
async def download(self, configuration):
    return await download_hls(self, configuration)

🛠️ Shared Utilities

base_api exports standardized parsing, sanitization, and helper functions used across scrapers:

📝 Logging & Cleanup

Configure logging once at application startup to capture provider and base API records in the same console or file output:

import logging
from base_api import configure_app_logging, DownloadFailed

configure_app_logging(log_file="api.log", level=logging.INFO)

With no logger_name, configure_app_logging() configures the root logger. Omit log_file for console output only. Log files append by default; overwrite_file=True truncates the selected file. The default format includes timestamp, logger name, level, filename, line number, function, and message. Logged exceptions include the original traceback and chained causes.

Contextual Logging (4.2.0)

Every library logger is wrapped in a ContextLogger adapter obtained via get_logger(name, owner=None). When operations run inside a log_context(owner, url) block, messages are automatically prefixed with [class=<class_name> url=<url>], and the context variables are injected into LogRecord.class_name and LogRecord.url. Because context is stored in ContextVar instances, concurrent downloads and background tasks offloaded via asyncio.to_thread keep their log context completely separate.

ERROR xvideos_api.api [class=xvideos_api.api.Video url=https://example.test/video/123] Download failed: disk full
Traceback (most recent call last):
  ...
OSError: disk full

Creating a client or core does not attach handlers or change your application's logging configuration. Without application setup, Python's default logging handler prints warnings and errors to stderr.

Provider CLI entry points configure INFO-level console logging automatically and log caught per-URL failures with tracebacks. Library imports do not configure the root logger; importing Tube8 or Thumbzilla no longer enables root DEBUG logging.

Logging a caught or stored exception

Use logger.exception() inside an except block. The APIs already log failures at their handling boundaries; add application logging when more context is needed. Outside the original handler, pass the stored exception explicitly:

logger = logging.getLogger(__name__)

try:
    await video.download(configuration=config)
except DownloadFailed as error:
    print(error.api, error.class_name, error.url)
    # The failure and its full traceback have already been logged.
except Exception:
    logger.exception("Application download failed for %s", video.url)
    raise

if not result.succeeded:
    error = result.error
    logger.error(
        "Scrape failed for %s: %s", result.url, error,
        exc_info=(type(error), error, error.__traceback__),
    )

exc_info=True alone cannot recover a traceback from a stored ScrapeResult. Ordinary download failures now raise DownloadFailed instead of returning False. Explicit cancellation via stop_event raises DownloadCancelled, and async task cancellation propagates asyncio.CancelledError without error logs. With DownloadConfigHLS(return_report=True), incomplete or cancelled downloads still return their explicit DownloadReport.

Core logging and cleanup

core.enable_logging() configures the core logger specifically. Use root logging above when the file should also receive provider and media-loader records. Optional remote logging uses log_ip/log_port on the core or http_ip/http_port on configure_app_logging().

import logging

core.enable_logging(level=logging.DEBUG)
core.enable_logging(log_file="api.log", level=logging.INFO)
core.enable_logging(
    log_ip="192.168.1.100",
    log_port=8080,
    level=logging.DEBUG,
)

# Always release the curl_cffi connection pool.
await core.close()

🚚 Migration from 3.x

🚨 Error Reference

Errors live in base_api.modules.errors; frequently used version 4 errors are also exported from base_api.

The common provider errors derive from ScraperException, which now derives from BaseScraperError. Catch common failures from base_api: DownloadFailed, NetworkError, NotFound, ProxyError, BotDetection, VideoUnavailable, LoginFailed, and RegionBlocked. Existing provider imports remain valid. Pornhub retains its own classes: its common errors subclass both the corresponding shared type and PornhubAPIError, preserving both ways of catching them.

Translated exceptions carry rich diagnostic attributes: error.url, error.class_name, and error.api. Provider request wrappers chain translated exceptions with raise ... from original_error, preserving lower-level causes in __cause__.

Download operations now raise DownloadFailed on failure instead of returning False. Download wrappers cover preparation failures (metadata loading, quality selection, output path setup); DownloadFailed includes the video URL and retains the original error in __cause__. An explicit DownloadCancelled or asyncio.CancelledError propagates without being wrapped as DownloadFailed and without error logs. With DownloadConfigHLS(return_report=True), incomplete or cancelled downloads still return their explicit DownloadReport.

📋 Changelog