[!WARNING] Version 4 deliberately removes the legacy BaseMedia.load(api=..., html=...) and TaskGroup-based Helper.iterator() contracts. Applications must migrate to source-aware media fields and the new scrape stream described below.

EAF Base API

What is this?

When using one of my Porn site APIs, you probably came across this package and wondered what it actually does, so here's a detailed answer.

A lot of Porn sites use very similar methods for m3u8 (HLS) parsing and other things. I also wanted to implement proxy support, and there was a lot of code that I would have rewritten in every API again and again. That's why I made this API package. The BaseCore class does all the necessary stuff like m3u8 parsing, a great caching system, network request fetching with retry attempts and proxy support.

Documentation (IMPORTANT!)

[!IMPORTANT] Configuring eaf_base_api is necessary if you use any of my Porn APIs, because they all depend on this project. Please read through the documentation to learn how PROXIES, CACHING and LOGGING etc... work!

You can find the documentation here ->: https://github.com/EchterAlsFake/API_Docs/blob/master/Porn_APIs/eaf_base_api.md

Source-aware media models

Use media_field() for every attribute populated by a remote loader. The first source is the highest-priority source if multiple sources provide the same field. Each configured loader is async and returns a complete mapping for all fields assigned to that source; loaders do not mutate the model directly.

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")
    available_qualities: list[int] | None = media_field("html")

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

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

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

Load exactly the information a caller needs:

video = Video(url=url, core=core)
await video.load_fields("title", "available_qualities")

# Or request a known source explicitly.
await video.load_sources("html")

# Convenience form that loads one field and returns it.
title = await video.get_field("title")

An unresolved field raises DataNotLoadedError with the exact field and eligible sources. A loader returning None marks the field as loaded and does not raise. Loader mappings are validated before any values are committed, preventing partial model updates after parser failures.

HTTP requests and caching

BaseCore exposes one method per response representation. Use the core as an async context manager so its connection pool is closed deterministically:

from base_api import BaseCore, CachePolicy

async with BaseCore() as core:
    response = await core.request("https://example.com/status")
    text = await core.fetch_text("https://example.com/page")
    data = await core.fetch_bytes("https://example.com/file")

    fresh_text = await core.fetch_text(
        "https://example.com/live",
        cache_policy=CachePolicy.REFRESH,
    )
    uncached_text = await core.fetch_text(
        "https://example.com/volatile",
        cache_policy=CachePolicy.BYPASS,
    )

Only successful GET text responses are cached. Cache keys distinguish parameters, request bodies, headers, and cookies without storing credentials in plaintext. CachePolicy.USE reads and writes the cache, REFRESH skips the read and replaces the entry, and BYPASS neither reads nor writes. Concurrent misses for the same request share one network operation.

Network failures and retryable HTTP statuses are retried automatically for idempotent methods. Set retry_non_idempotent=True only when repeating a POST or PATCH is known to be safe.

Concurrent page and media iteration

Helper uses bounded asyncio task sets. Completion order is the default because it exposes fast media without waiting for slower earlier media. Original page and extractor order is available with ResultOrder.ORIGINAL.

from base_api import Helper, ResultOrder
from base_api.modules.config import IteratorConfig

helper = Helper(core=core, constructor=Video)
stream = helper.iterator(
    page_urls,
    extractor_videos,
    iterator_config=IteratorConfig(
        max_page_concurrency=3,
        max_item_concurrency=20,
        load_specific_fields=("title", "available_qualities"),
        order=ResultOrder.COMPLETION,  # The default.
    ),
)

# The context manager guarantees immediate task cleanup if this loop breaks early.
async with stream:
    async for result in stream:
        if not result.succeeded:
            logger.error("%s failed: %s", result.stage, result.error)
            continue
        video = result.unwrap()

Use IteratorConfig(order=ResultOrder.ORIGINAL) when presentation order matters. Page and item failures independently support ErrorMode.YIELD, ErrorMode.SKIP, or ErrorMode.RAISE. RetryPolicy provides a strict maximum attempt count and optional exponential delay; the independent page and item handlers return an ErrorAction and cannot create an unbounded retry loop.

Can I use this for myself?

Yes, you can, but I may change stuff here and there from time to time, and it would maybe break your project. I would not recommend you to install and use it as a package, but just copy the code you need.

I can recommend everyone the download functions for HLS streaming since, for example, the threaded preset is very well optimized. If you just use mine, you need to consume less caffeine and brain cells to make such a function :)

License

Licensed under The AGPLv3 license.
Copyright (C) 2024-2026 Johannes Habel

Project links

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Newer release available (4.1.1)

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for EchterAlsFake from gravatar.com EchterAlsFake

Credits

PyPI data
Data sourced directly from PyPI's database.

Author: Johannes Habel

License expression

AGPL-3.0-or-later
View SPDX License List

Requires

Python >=3.12

Provides Extra

hls

Classifiers

Report project as malware

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

eaf_base_api-4.1.0.tar.gz (61.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

eaf_base_api-4.1.0-py3-none-any.whl (63.4 kB view details)

Uploaded Python 3

File details

Details for the file eaf_base_api-4.1.0.tar.gz.

File metadata

  • Download URL: eaf_base_api-4.1.0.tar.gz
  • Upload date:
  • Size: 61.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.10.7 {"installer":{"name":"uv","version":"0.10.7","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Arch Linux","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

File hashes

Hashes for eaf_base_api-4.1.0.tar.gz
Algorithm Hash digest
SHA256 d24c92bbae49b89a05f68531f9d4adbd05cbb3130050737d35ed04bb0d908992
MD5 8141d1563bb60e94398d19aae037fbf7
BLAKE2b-256 c5f0f6ab37e84a9276750cc1dcb778cbe689b8ee5195c8f5847048d58e99aa86

See more details on using hashes here.

File details

Details for the file eaf_base_api-4.1.0-py3-none-any.whl.

File metadata

  • Download URL: eaf_base_api-4.1.0-py3-none-any.whl
  • Upload date:
  • Size: 63.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.10.7 {"installer":{"name":"uv","version":"0.10.7","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Arch Linux","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

File hashes

Hashes for eaf_base_api-4.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 1c5df058a5b0ecd2856f7c3db223921691d629e4262b0e3ec8356dc3d595d13f
MD5 c5ee33b737fa97e595d36e72ed9aaceb
BLAKE2b-256 45b67d0d62c595386e9d5f098f783c37d6b4f990cb2b214e476a31d9076e7b7d

See more details on using hashes here.

Release history Release notifications | RSS feed

4.1.1

2 files

This release

4.1.0 This release

2 files

4.0.1

2 files

4.0.0

2 files

3.3.5

2 files

3.3.4

2 files

3.3.3

2 files

3.3.2

2 files

3.3.1

2 files

3.3

2 files

3.2.4

2 files

3.2.3

2 files

3.2.2

2 files

3.2.1

2 files

3.2

2 files

3.1

2 files

3.0

2 files

2.5.4

2 files

2.5.3

2 files

2.5.2

2 files

2.5.1

2 files

2.5.0

2 files

2.4.9

2 files

2.4.8

2 files

2.4.7

2 files

2.4.6

2 files

2.4.5

2 files

2.4.4

2 files

2.4.3

2 files

2.4.2

2 files

2.4.0

2 files

2.3.8

2 files

2.3.7

2 files

2.3.6

2 files

2.3.5

2 files

2.3.4

2 files

2.3.3

2 files

2.3.2

2 files

2.2.9

2 files

2.2.8

2 files

2.2.7

2 files

2.2.6

2 files

2.2.5

2 files

2.2.4

2 files

2.2.3

2 files

2.2.2

2 files

2.2.1

2 files

2.2

2 files

2.1.1

2 files

2.1

2 files

2.0

2 files

1.6.3

2 files

1.6.2

2 files

1.6.1

2 files

1.6

2 files

1.5

2 files

1.3

2 files

1.2

2 files

1.1

2 files

1.0

2 files

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page