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, 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 Binding

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"

# 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.

💾 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. Exhausting it cleans up normally; use it as an async context manager whenever the consumer may stop early.

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()  # Same object as result.item after the check

⬇️ 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.

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.

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.

📝 Logging & Cleanup

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.

📋 Changelog