Utils

persista.utils

Contain utilities.

persista.utils.asyncio

Contain asyncio utility classes.

persista.utils.asyncio.EmptyAsyncIterator

An async iterator that is always exhausted.

Useful for implementing async stores/iterables that never yield anything (e.g. a no-op store), without resorting to an async generator whose yield line can never be reached.

Example

>>> import asyncio
>>> from persista.utils.asyncio import EmptyAsyncIterator
>>> async def main():
...     return [x async for x in EmptyAsyncIterator()]
...
>>> asyncio.run(main())
[]

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()

persista.utils.path

Contain path utility functions.

persista.utils.path.prepare_store_path

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

Prepare a path for use as a file-backed store's location.

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 the underlying backend can create the database file without failing on a missing directory.

Parameters:

Name	Type	Description	Default
path	Path | str	The 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.

Example

>>> import tempfile
>>> from persista.utils.path import prepare_store_path
>>> prepare_store_path(":memory:")
':memory:'
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = prepare_store_path(f"{tmpdir}/nested/data.db")
...     path.parent.is_dir()
...
True