Utils

persista.utils ¶

Contain utilities.

persista.utils.duckdb ¶

Contain DuckDB utility functions.

persista.utils.duckdb.prepare_duckdb_path ¶

prepare_duckdb_path(path: Path | str) -> Path | str

Prepare a path for use with duckdb.connect.

If path is the special in-memory sentinel (":memory:"), it is returned unchanged. Otherwise, path is sanitized and its parent directory is created if it does not already exist, so that DuckDB can create the database file without failing on a missing directory.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The DuckDB connection target -- either the in-memory sentinel `":memory:"`, or a filesystem path to a database file.	required

Returns:

Type	Description
`Path \| str`	`":memory:"` unchanged, or the sanitized, absolute `Path`
`Path \| str`	whose parent directory is guaranteed to exist.

persista.utils.http_httpx ¶

Provide HTTP helper functions for fetching remote content using httpx.

persista.utils.http_httpx.fetch_response ¶

fetch_response(
    url: str,
    timeout: int = 30,
    max_retries: int = 3,
    headers: dict[str, str] | None = None,
    retry_status_codes: (
        set[int] | frozenset[int]
    ) = DEFAULT_RETRY_STATUS_CODES,
    client: Client | None = None,
) -> Response

Fetch a URL with automatic retries and timeout.

Uses exponential backoff to handle transient network failures, connection timeouts, and 5xx server errors. Successive retry delays are 1s, 2s, 4s, and so on up to max_retries attempts.

If a client is provided it is used directly, allowing callers to share a single client across multiple calls for connection pooling. Otherwise a new client is created and closed automatically.

Parameters:

Name	Type	Description	Default
`url`	`str`	The full URL to fetch.	required
`timeout`	`int`	Request timeout in seconds per attempt. Defaults to 30. Ignored when `client` is provided.	`30`
`max_retries`	`int`	Maximum number of retry attempts on transient failures. Defaults to 3. Set to 0 to disable retries.	`3`
`headers`	`dict[str, str] \| None`	HTTP headers to include in the request. Pass `None` to send no custom headers (the default). Pass an empty dict to send no headers explicitly.	`None`
`retry_status_codes`	`set[int] \| frozenset[int]`	The HTTP status codes that trigger a retry. Defaults to `{429, 500, 502, 503, 504}`.	`DEFAULT_RETRY_STATUS_CODES`
`client`	`Client \| None`	An optional :class:`httpx.Client` to reuse. When `None`, a new client is created and closed after the request completes.	`None`

Returns:

Name	Type	Description
`The`	`Response`	class:`httpx.Response` object for the completed request.

Raises:

Type	Description
`RuntimeError`	if the `httpx` package is not installed.
`HTTPStatusError`	On 4xx/5xx responses that are not retried (e.g. 404, 403).
`TransportError`	If the host is unreachable or the request times out after all retries are exhausted.

Example

>>> from persista.utils.http_httpx import fetch_response
>>> response = fetch_response(  # doctest: +SKIP
...     "https://jsonplaceholder.typicode.com/todos/1",
...     timeout=10,
...     max_retries=5,
... )

persista.utils.http_httpx.fetch_response_async `async` ¶

fetch_response_async(
    url: str,
    timeout: int = 30,
    max_retries: int = 3,
    headers: dict[str, str] | None = None,
    retry_status_codes: (
        set[int] | frozenset[int]
    ) = DEFAULT_RETRY_STATUS_CODES,
    client: AsyncClient | None = None,
) -> Response

Fetch a URL asynchronously with automatic retries and timeout.

Uses exponential backoff to handle transient network failures, connection timeouts, and 5xx server errors. Successive retry delays are 1s, 2s, 4s, and so on up to max_retries attempts.

If a client is provided it is used directly, allowing callers to share a single client across multiple calls for connection pooling. Otherwise a new client is created and closed automatically.

Parameters:

Name	Type	Description	Default
`url`	`str`	The full URL to fetch.	required
`timeout`	`int`	Request timeout in seconds per attempt. Defaults to 30. Ignored when `client` is provided.	`30`
`max_retries`	`int`	Maximum number of retry attempts on transient failures. Defaults to 3. Set to 0 to disable retries.	`3`
`headers`	`dict[str, str] \| None`	HTTP headers to include in the request. Pass `None` to send no custom headers (the default). Pass an empty dict to send no headers explicitly.	`None`
`retry_status_codes`	`set[int] \| frozenset[int]`	The HTTP status codes that trigger a retry. Defaults to `{429, 500, 502, 503, 504}`.	`DEFAULT_RETRY_STATUS_CODES`
`client`	`AsyncClient \| None`	An optional :class:`httpx.AsyncClient` to reuse. When `None`, a new client is created and closed after the request completes.	`None`

Returns:

Name	Type	Description
`The`	`Response`	class:`httpx.Response` object for the completed request.

Raises:

Type	Description
`RuntimeError`	if the `httpx` package is not installed.
`HTTPStatusError`	On 4xx/5xx responses that are not retried (e.g. 404, 403).
`TransportError`	If the host is unreachable or the request times out after all retries are exhausted.

Example

>>> import asyncio
>>> from persista.utils.http_httpx import fetch_response_async
>>> response = asyncio.run(  # doctest: +SKIP
...     fetch_response_async(
...         "https://jsonplaceholder.typicode.com/todos/1",
...         timeout=10,
...         max_retries=5,
...     )
... )

persista.utils.http_requests ¶

Provide HTTP helper functions for fetching remote content.

persista.utils.http_requests.create_session ¶

create_session(
    max_retries: int = 3,
    retry_status_codes: list[int] | None = None,
    backoff_factor: float = 1,
) -> Session

Create a :class:requests.Session with a retry adapter mounted.

Configures exponential backoff retries for transient failures on both https:// and http:// connections. Useful for sharing a single session across multiple requests for connection pooling.

Parameters:

Name	Type	Description	Default
`max_retries`	`int`	Maximum number of retry attempts on transient failures with exponential backoff. Defaults to 3.	`3`
`retry_status_codes`	`list[int] \| None`	HTTP status codes that should trigger a retry. Pass `None` to use the default set (429, 500, 502, 503, 504).	`None`
`backoff_factor`	`float`	Multiplier used to compute the delay between retry attempts. Successive delays are `backoff_factor * (2 ** (retry_number - 1))` seconds. Defaults to 1.	`1`

Returns:

Type	Description
`Session`	A configured :class:`requests.Session` with retry adapters mounted
`Session`	on both `https://` and `http://`.

Example

>>> from persista.utils.http_requests import create_session
>>> session = create_session(max_retries=5)

persista.utils.http_requests.fetch_response ¶

fetch_response(
    url: str,
    timeout: int = 30,
    max_retries: int = 3,
    retry_status_codes: list[int] | None = None,
    backoff_factor: float = 1,
    headers: dict[str, str] | None = None,
    session: Session | None = None,
) -> Response

Fetch a URL with automatic retries and timeout.

Uses exponential backoff to handle transient network failures, connection timeouts, and 5xx server errors. Successive retry delays are backoff_factor * 1s, 2s, 4s, and so on up to max_retries attempts.

If a session is provided it is used directly, allowing callers to share a single session across multiple calls for connection pooling. Otherwise a new session is created and closed automatically.

Parameters:

Name	Type	Description	Default
`url`	`str`	The full URL to fetch.	required
`timeout`	`int`	Request timeout in seconds per attempt. Defaults to 30.	`30`
`max_retries`	`int`	Maximum number of retry attempts on transient failures. Defaults to 3. Set to 0 to disable retries. Ignored when `session` is provided.	`3`
`retry_status_codes`	`list[int] \| None`	HTTP status codes that should trigger a retry. Pass `None` to use the default set (429, 500, 502, 503, 504). Ignored when `session` is provided.	`None`
`backoff_factor`	`float`	Multiplier used to compute the delay between retry attempts. Defaults to 1. Ignored when `session` is provided.	`1`
`headers`	`dict[str, str] \| None`	HTTP headers to include in the request. Pass `None` to send no custom headers (the default). Pass an empty dict to send no headers explicitly.	`None`
`session`	`Session \| None`	An optional :class:`requests.Session` to reuse. When `None`, a new session is created via :func:`create_session` and closed after the request completes.	`None`

Returns:

Name	Type	Description
`The`	`Response`	class:`requests.Response` object for the completed request.

Raises:

Type	Description
`RuntimeError`	if the `requests` package is not installed.
`ConnectTimeout`	If all retry attempts exceed `timeout` seconds.
`HTTPError`	On 4xx/5xx responses that are not retried (e.g. 404, 403).
`ConnectionError`	If the host is unreachable after all retries are exhausted.
`RequestException`	For any other unrecoverable network failure.

Example

>>> from persista.utils.http_requests import fetch_response
>>> html = fetch_response(  # doctest: +SKIP
...     "https://jsonplaceholder.typicode.com/todos/1",
...     timeout=10,
...     max_retries=5,
... )

persista.utils.imports ¶

Contain utilities for optional dependencies.

persista.utils.imports.aiosqlite_available ¶

aiosqlite_available(fn: F) -> F

Implement a decorator to execute a function only if aiosqlite package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `aiosqlite` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import aiosqlite_available
>>> @aiosqlite_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.check_aiosqlite ¶

check_aiosqlite() -> None

Check if the aiosqlite package is installed.

Raises:

Type	Description
`RuntimeError`	if the `aiosqlite` package is not installed.

Example

>>> from persista.utils.imports import check_aiosqlite
>>> check_aiosqlite()

persista.utils.imports.check_duckdb ¶

check_duckdb() -> None

Check if the duckdb package is installed.

Raises:

Type	Description
`RuntimeError`	if the `duckdb` package is not installed.

Example

>>> from persista.utils.imports import check_duckdb
>>> check_duckdb()

persista.utils.imports.check_faker ¶

check_faker() -> None

Check if the faker package is installed.

Raises:

Type	Description
`RuntimeError`	if the `faker` package is not installed.

Example

>>> from persista.utils.imports import check_faker
>>> check_faker()

persista.utils.imports.check_httpx ¶

check_httpx() -> None

Check if the httpx package is installed.

Raises:

Type	Description
`RuntimeError`	if the `httpx` package is not installed.

Example

>>> from persista.utils.imports import check_httpx
>>> check_httpx()

persista.utils.imports.check_lmdb ¶

check_lmdb() -> None

Check if the lmdb package is installed.

Raises:

Type	Description
`RuntimeError`	if the `lmdb` package is not installed.

Example

>>> from persista.utils.imports import check_lmdb
>>> check_lmdb()

persista.utils.imports.check_psycopg ¶

check_psycopg() -> None

Check if the psycopg package is installed.

Raises:

Type	Description
`RuntimeError`	if the `psycopg` package is not installed.

Example

>>> from persista.utils.imports import check_psycopg
>>> check_psycopg()

persista.utils.imports.check_redis ¶

check_redis() -> None

Check if the redis package is installed.

Raises:

Type	Description
`RuntimeError`	if the `redis` package is not installed.

Example

>>> from persista.utils.imports import check_redis
>>> check_redis()

persista.utils.imports.check_requests ¶

check_requests() -> None

Check if the requests package is installed.

Raises:

Type	Description
`RuntimeError`	if the `requests` package is not installed.

Example

>>> from persista.utils.imports import check_requests
>>> check_requests()

persista.utils.imports.check_urllib3 ¶

check_urllib3() -> None

Check if the urllib3 package is installed.

Raises:

Type	Description
`RuntimeError`	if the `urllib3` package is not installed.

Example

>>> from persista.utils.imports import check_urllib3
>>> check_urllib3()

persista.utils.imports.duckdb_available ¶

duckdb_available(fn: F) -> F

Implement a decorator to execute a function only if duckdb package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `duckdb` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import duckdb_available
>>> @duckdb_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.faker_available ¶

faker_available(fn: F) -> F

Implement a decorator to execute a function only if faker package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `faker` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import faker_available
>>> @faker_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.httpx_available ¶

httpx_available(fn: F) -> F

Implement a decorator to execute a function only if httpx package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `httpx` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import httpx_available
>>> @httpx_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.is_aiosqlite_available `cached` ¶

is_aiosqlite_available() -> bool

Indicate if the aiosqlite package is installed or not.

Returns:

Type	Description
`bool`	`True` if `aiosqlite` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_aiosqlite_available
>>> is_aiosqlite_available()

persista.utils.imports.is_duckdb_available `cached` ¶

is_duckdb_available() -> bool

Indicate if the duckdb package is installed or not.

Returns:

Type	Description
`bool`	`True` if `duckdb` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_duckdb_available
>>> is_duckdb_available()

persista.utils.imports.is_faker_available `cached` ¶

is_faker_available() -> bool

Indicate if the faker package is installed or not.

Returns:

Type	Description
`bool`	`True` if `faker` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_faker_available
>>> is_faker_available()

persista.utils.imports.is_httpx_available `cached` ¶

is_httpx_available() -> bool

Indicate if the httpx package is installed or not.

Returns:

Type	Description
`bool`	`True` if `httpx` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_httpx_available
>>> is_httpx_available()

persista.utils.imports.is_lmdb_available `cached` ¶

is_lmdb_available() -> bool

Indicate if the lmdb package is installed or not.

Returns:

Type	Description
`bool`	`True` if `lmdb` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_lmdb_available
>>> is_lmdb_available()

persista.utils.imports.is_psycopg_available `cached` ¶

is_psycopg_available() -> bool

Indicate if the psycopg package is installed or not.

Returns:

Type	Description
`bool`	`True` if `psycopg` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_psycopg_available
>>> is_psycopg_available()

persista.utils.imports.is_redis_available `cached` ¶

is_redis_available() -> bool

Indicate if the redis package is installed or not.

Returns:

Type	Description
`bool`	`True` if `redis` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_redis_available
>>> is_redis_available()

persista.utils.imports.is_requests_available `cached` ¶

is_requests_available() -> bool

Indicate if the requests package is installed or not.

Returns:

Type	Description
`bool`	`True` if `requests` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_requests_available
>>> is_requests_available()

persista.utils.imports.is_urllib3_available `cached` ¶

is_urllib3_available() -> bool

Indicate if the urllib3 package is installed or not.

Returns:

Type	Description
`bool`	`True` if `urllib3` is available otherwise `False`.

Example

>>> from persista.utils.imports import is_urllib3_available
>>> is_urllib3_available()

persista.utils.imports.lmdb_available ¶

lmdb_available(fn: F) -> F

Implement a decorator to execute a function only if lmdb package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `lmdb` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import lmdb_available
>>> @lmdb_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.psycopg_available ¶

psycopg_available(fn: F) -> F

Implement a decorator to execute a function only if psycopg package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `psycopg` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import psycopg_available
>>> @psycopg_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.raise_aiosqlite_missing_error ¶

raise_aiosqlite_missing_error() -> NoReturn

Raise a RuntimeError to indicate the aiosqlite package is missing.

persista.utils.imports.raise_duckdb_missing_error ¶

raise_duckdb_missing_error() -> NoReturn

Raise a RuntimeError to indicate the duckdb package is missing.

persista.utils.imports.raise_faker_missing_error ¶

raise_faker_missing_error() -> NoReturn

Raise a RuntimeError to indicate the faker package is missing.

persista.utils.imports.raise_httpx_missing_error ¶

raise_httpx_missing_error() -> NoReturn

Raise a RuntimeError to indicate the httpx package is missing.

persista.utils.imports.raise_lmdb_missing_error ¶

raise_lmdb_missing_error() -> NoReturn

Raise a RuntimeError to indicate the lmdb package is missing.

persista.utils.imports.raise_psycopg_missing_error ¶

raise_psycopg_missing_error() -> NoReturn

Raise a RuntimeError to indicate the psycopg package is missing.

persista.utils.imports.raise_redis_missing_error ¶

raise_redis_missing_error() -> NoReturn

Raise a RuntimeError to indicate the redis package is missing.

persista.utils.imports.raise_requests_missing_error ¶

raise_requests_missing_error() -> NoReturn

Raise a RuntimeError to indicate the requests package is missing.

persista.utils.imports.raise_urllib3_missing_error ¶

raise_urllib3_missing_error() -> NoReturn

Raise a RuntimeError to indicate the urllib3 package is missing.

persista.utils.imports.redis_available ¶

redis_available(fn: F) -> F

Implement a decorator to execute a function only if redis package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `redis` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import redis_available
>>> @redis_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.requests_available ¶

requests_available(fn: F) -> F

Implement a decorator to execute a function only if requests package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `requests` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import requests_available
>>> @requests_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

persista.utils.imports.urllib3_available ¶

urllib3_available(fn: F) -> F

Implement a decorator to execute a function only if urllib3 package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `urllib3` package is installed, otherwise `None`.

Example

>>> from persista.utils.imports import urllib3_available
>>> @urllib3_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

Utils

persista.utils ¶

persista.utils.duckdb ¶

persista.utils.duckdb.prepare_duckdb_path ¶

persista.utils.http_httpx ¶

persista.utils.http_httpx.fetch_response ¶

persista.utils.http_httpx.fetch_response_async async ¶

persista.utils.http_requests ¶

persista.utils.http_requests.create_session ¶

persista.utils.http_requests.fetch_response ¶

persista.utils.imports ¶

persista.utils.imports.aiosqlite_available ¶

persista.utils.imports.check_aiosqlite ¶

persista.utils.imports.check_duckdb ¶

persista.utils.imports.check_faker ¶

persista.utils.imports.check_httpx ¶

persista.utils.imports.check_lmdb ¶

persista.utils.imports.check_psycopg ¶

persista.utils.imports.check_redis ¶

persista.utils.imports.check_requests ¶

persista.utils.imports.check_urllib3 ¶

persista.utils.imports.duckdb_available ¶

persista.utils.imports.faker_available ¶

persista.utils.imports.httpx_available ¶

persista.utils.imports.is_aiosqlite_available cached ¶

persista.utils.imports.is_duckdb_available cached ¶

persista.utils.imports.is_faker_available cached ¶

persista.utils.imports.is_httpx_available cached ¶

persista.utils.imports.is_lmdb_available cached ¶

persista.utils.imports.is_psycopg_available cached ¶

persista.utils.imports.is_redis_available cached ¶

persista.utils.imports.is_requests_available cached ¶

persista.utils.imports.is_urllib3_available cached ¶

persista.utils.imports.lmdb_available ¶

persista.utils.imports.psycopg_available ¶

persista.utils.imports.raise_aiosqlite_missing_error ¶

persista.utils.imports.raise_duckdb_missing_error ¶

persista.utils.imports.raise_faker_missing_error ¶

persista.utils.imports.raise_httpx_missing_error ¶

persista.utils.imports.raise_lmdb_missing_error ¶

persista.utils.imports.raise_psycopg_missing_error ¶

persista.utils.imports.raise_redis_missing_error ¶

persista.utils.imports.raise_requests_missing_error ¶

persista.utils.imports.raise_urllib3_missing_error ¶

persista.utils.imports.redis_available ¶

persista.utils.imports.requests_available ¶

persista.utils.imports.urllib3_available ¶

persista.utils.http_httpx.fetch_response_async `async` ¶

persista.utils.imports.is_aiosqlite_available `cached` ¶

persista.utils.imports.is_duckdb_available `cached` ¶

persista.utils.imports.is_faker_available `cached` ¶

persista.utils.imports.is_httpx_available `cached` ¶

persista.utils.imports.is_lmdb_available `cached` ¶

persista.utils.imports.is_psycopg_available `cached` ¶

persista.utils.imports.is_redis_available `cached` ¶

persista.utils.imports.is_requests_available `cached` ¶

persista.utils.imports.is_urllib3_available `cached` ¶