vyra_base.storage package

Submodules

vyra_base.storage.db_access module

class vyra_base.storage.db_access.DBSTATUS(*values)

Bases: str, Enum

Enumeration of database operation status codes.

Variables:

• SUCCESS – Operation completed successfully.

• ERROR – An error occurred during the operation.

• NOT_FOUND – Requested resource was not found.

• NOT_ALLOWED – Operation is not allowed.

• NOT_AUTHORIZED – User is not authorized for this operation.

• CONFLICT – Operation conflicts with existing data.

SUCCESS = 'success'

ERROR = 'error'

NOT_FOUND = 'not_found'

NOT_ALLOWED = 'not_allowed'

NOT_AUTHORIZED = 'not_authorized'

CONFLICT = 'conflict'

class vyra_base.storage.db_access.DBTYPE(*values)

Bases: str, Enum

Enumeration of supported database types.

Variables:

• SQLITE – SQLite database engine.

• MYSQL – MySQL database engine.

• POSTGRESQL – PostgreSQL database engine.

SQLITE = 'sqlite'

MYSQL = 'mysql'

POSTGRESQL = 'postgresql'

class vyra_base.storage.db_access.DBMESSAGE

Bases: object

Collection of standard database error messages.

Variables:

DEFAULT_ERROR – Generic error message for database operations.

DEFAULT_ERROR = 'Something went wrong while processing the query. See the details.'

class vyra_base.storage.db_access.DbAccess(module_name, db_config_path=None, db_config=None, db_type=DBTYPE.SQLITE)

Bases: Storage

Baseclass for database access.

Parameters:

• module_name (str)

• db_config_path (str)

• db_config (dict)

• db_type (DBTYPE)

__init__(module_name, db_config_path=None, db_config=None, db_type=DBTYPE.SQLITE)

Initialize database object.

Parameters:

• module_name (str) – The id of the V.Y.R.A. module.

• db_config_path (str) – Path to the ini file of your sqlalchemy config. Defaults to WORKING_PATH+PATH_DB_CONFIG.

• db_config (dict) – Dictionary with database configuration.

• db_type (DBTYPE)

Raises:

ValueError – If neither db_config_path nor db_config is provided, or if db_config is not a dict.

Return type:

None

db_engine: AsyncEngine

session()

Create a session for the database.

Returns:

A session object.

Return type:

Union[sessionmaker, async_sessionmaker]

async create_all_tables()

Create all database tables that are children of the SQLAlchemy Base class.

This method will create all tables that are defined in the SQLAlchemy Base class. It will also create the tables if they do not exist. This method will not create the tables if they already exist.

Returns:

Status of the operation.

Return type:

str

async create_selected_table(table_structs)

Create new database table.

Parameters:

table_structs (list[Base]) – Table configurations as python classes (SQLAlchemy declarative base style).

Returns:

Status of the operation.

Return type:

str

async drop_table(table)

Delete table from database.

Parameters:

table (Base) – Table class (SQLAlchemy declarative base).

Returns:

Status of the operation.

Return type:

str

async check_table_exists(table)

Check if a table exists in the database.

Parameters:

table (Type[Base]) – Table class (SQLAlchemy declarative base).

Returns:

True if the table exists, False otherwise.

Return type:

bool

vyra_base.storage.db_manipulator module

Handling database datatables

class vyra_base.storage.db_manipulator.DBReturnValue(status=None, value='', details='')

Bases: object

Standardized return value container for database operations.

Encapsulates the result of database operations with status, value, and details. Provides convenience methods for setting error/success states.

Variables:

• status – Operation status (see DBSTATUS enum).

• value – Main return value (data or error message).

• details – Additional details about the operation.

Parameters:

• status (Union[str, None])

• value (Union[bool, str, list, dict])

• details (Union[bool, str, list, dict])

__init__(status=None, value='', details='')

Parameters:

• status (None | str)

• value (bool | str | list | dict)

• details (bool | str | list | dict)

status: str | None = None

value: bool | str | list | dict = ''

details: bool | str | list | dict = ''

error_return(details='')

Set the return value to error status.

Parameters:

details (Union[bool, str, list, dict]) – Additional error details.

Returns:

Self with error status set.

Return type:

DBReturnValue

success_return()

Set the return value to success status.

Returns:

Self with success status set.

Return type:

DBReturnValue

class vyra_base.storage.db_manipulator.DbManipulator(**kwargs)

Bases: object

Datatable class manipulator

__init__(**kwargs)

Wrapper for synchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

get_table_structure()

Read the datatable structure from the config file.

Returns:

Table structure information.

Return type:

DBReturnValue

async get_by_id(id=-1)

Read line from datatable of database by ‘id’.

If the id is -1 or None, the last line will be read.

Parameters:

id (Union[UUID, int]) – Private key of the table element to select the row to be read.

Returns:

Row data or not found status.

Return type:

DBReturnValue

async get_all(filters=None, order_by=None, limit=None)

Read all lines from datatable of database.

Parameters:

• filters (dict) – Filter elements to identify the rows to be read.

• order_by (str, optional) – Column name to order the result.

• limit (int, optional) – Number of lines to be read.

Returns:

List of rows or not found status.

Return type:

DBReturnValue

async update(data, filters=None)

Update a line in a datatable of database by given data.

If the table config has a field ‘max_lines’ and the number of lines are greater than this field, the lowest ‘id’ value will be deleted and the new entry will be added by a incremented ‘id’ value.

Parameters:

• data (dict) – Update data in an existing entry.

• filters (Optional[dict]) – Filter to select the row to be updated.

Returns:

Update status.

Return type:

DBReturnValue

async add(data)

Add a new a row in a datatable.

Parameters:

data (dict) – Content of the new entry to be added to the table.

Returns:

Add status and details.

Return type:

DBReturnValue

async get_one(filters=None)

Read a single row from the database table, optionally filtered.

Returns the first matching model instance using scalars().first(), consistent with get_all(). Use this instead of get_by_id when you need to filter by arbitrary columns or want a guaranteed model instance (not a raw Row).

Parameters:

filters (dict | None) – Optional column/value pairs to filter by.

Returns:

DBReturnValue with .value = model instance or None.

Return type:

DBReturnValue

async add_instance(obj)

Add a pre-built model instance to the database.

Use this instead of add() when you need full control over the object’s fields (e.g. custom primary keys or enum values) and have already constructed the model instance yourself.

Parameters:

obj (Any) – A model instance (subclass of the declarative base).

Returns:

Add status and details.

Return type:

DBReturnValue

async delete(id)

Update a line in a datatable of database by a given ‘id’.

Parameters:

id (Any) – Private key of the table element to select the row to be deleted.

Returns:

Delete status and details.

Return type:

DBReturnValue

async bulk_add(data)

Add multiple rows to a datatable.

Parameters:

data (list[dict]) – Data list to be added to the table.

Returns:

Bulk add status and details.

Return type:

DBReturnValue

async bulk_delete(filters)

Delete multiple rows in a datatable by given filters.

Parameters:

filters (dict) – Filter elements to identify the rows to be deleted.

Returns:

Bulk delete status and details.

Return type:

DBReturnValue

async exists(id)

Check if a row exists in the datatable by a given ‘id’.

Parameters:

id (int) – Private key of the table element to check.

Returns:

Existence status.

Return type:

DBReturnValue

async count(filters)

Count the number of rows in a datatable matching given filters.

Parameters:

filters (dict) – Filter elements to identify the rows to be counted.

Returns:

Number of matching rows.

Return type:

DBReturnValue

to_dict(obj, exclude=None)

Convert a SQLAlchemy model instance to a dictionary.

Parameters:

• obj (Base) – SQLAlchemy model instance.

• exclude (list, optional) – List of column names to exclude.

Returns:

Dictionary representation of the object.

Return type:

dict

vyra_base.storage.storage module

class vyra_base.storage.storage.Storage

Bases: object

Base class for storage access.

This class provides a common interface for different storage backends.

It can be used to access different storage backends like databases, files, etc.

vyra_base.storage.tb_base module

class vyra_base.storage.tb_base.Base(**kwargs)

Bases: AsyncAttrs, DeclarativeBase

Base class for all SQLAlchemy models in the application.

Parameters:

kwargs (Any)

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

Parameters:

• self (Any)

• kwargs (Any)

Return type:

None

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[registry] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class vyra_base.storage.tb_base.IntEnum(enumtype, *args, **kwargs)

Bases: TypeDecorator

A SQLAlchemy TypeDecorator for storing Python IntEnum values as integers in the database.

This class allows seamless conversion between Python IntEnum members and their integer representation in the database. When binding parameters, it converts IntEnum members to their integer values. When retrieving results, it converts integers back to the corresponding IntEnum members.

Parameters:

enumtype (Type[IntEnum]) – The IntEnum class to use for conversion.

impl

alias of Integer

__init__(enumtype, *args, **kwargs)

Construct a TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute ‘as-is’, ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

process_bind_param(value, dialect)

Convert Python enum to database value before storing.

Parameters:

• value – Python enum value to convert.

• dialect – SQLAlchemy dialect.

Returns:

Integer value for database storage.

process_result_value(value, dialect)

Convert database value back to Python enum after retrieval.

Parameters:

• value – Database integer value.

• dialect – SQLAlchemy dialect.

Returns:

Python enum instance.

Module contents

VYRA Base Storage Module

Provides access to SQLite database and Redis storage.

Public API for external developers

Important: All database tables MUST use the tb_ prefix! Example: tb_parameters, tb_sensor_data, tb_logs

class vyra_base.storage.Base(**kwargs)

Bases: AsyncAttrs, DeclarativeBase

Base class for all SQLAlchemy models in the application.

Parameters:

kwargs (Any)

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

Parameters:

• self (Any)

• kwargs (Any)

Return type:

None

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[registry] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class vyra_base.storage.IntEnum(enumtype, *args, **kwargs)

Bases: TypeDecorator

A SQLAlchemy TypeDecorator for storing Python IntEnum values as integers in the database.

This class allows seamless conversion between Python IntEnum members and their integer representation in the database. When binding parameters, it converts IntEnum members to their integer values. When retrieving results, it converts integers back to the corresponding IntEnum members.

Parameters:

enumtype (Type[IntEnum]) – The IntEnum class to use for conversion.

impl

alias of Integer

__init__(enumtype, *args, **kwargs)

Construct a TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute ‘as-is’, ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

process_bind_param(value, dialect)

Convert Python enum to database value before storing.

Parameters:

• value – Python enum value to convert.

• dialect – SQLAlchemy dialect.

Returns:

Integer value for database storage.

process_result_value(value, dialect)

Convert database value back to Python enum after retrieval.

Parameters:

• value – Database integer value.

• dialect – SQLAlchemy dialect.

Returns:

Python enum instance.

class vyra_base.storage.DbAccess(module_name, db_config_path=None, db_config=None, db_type=DBTYPE.SQLITE)

Bases: Storage

Baseclass for database access.

Parameters:

• module_name (str)

• db_config_path (str)

• db_config (dict)

• db_type (DBTYPE)

__init__(module_name, db_config_path=None, db_config=None, db_type=DBTYPE.SQLITE)

Initialize database object.

Parameters:

• module_name (str) – The id of the V.Y.R.A. module.

• db_config_path (str) – Path to the ini file of your sqlalchemy config. Defaults to WORKING_PATH+PATH_DB_CONFIG.

• db_config (dict) – Dictionary with database configuration.

• db_type (DBTYPE)

Raises:

ValueError – If neither db_config_path nor db_config is provided, or if db_config is not a dict.

Return type:

None

db_engine: AsyncEngine

session()

Create a session for the database.

Returns:

A session object.

Return type:

Union[sessionmaker, async_sessionmaker]

async create_all_tables()

Create all database tables that are children of the SQLAlchemy Base class.

This method will create all tables that are defined in the SQLAlchemy Base class. It will also create the tables if they do not exist. This method will not create the tables if they already exist.

Returns:

Status of the operation.

Return type:

str

async create_selected_table(table_structs)

Create new database table.

Parameters:

table_structs (list[Base]) – Table configurations as python classes (SQLAlchemy declarative base style).

Returns:

Status of the operation.

Return type:

str

async drop_table(table)

Delete table from database.

Parameters:

table (Base) – Table class (SQLAlchemy declarative base).

Returns:

Status of the operation.

Return type:

str

async check_table_exists(table)

Check if a table exists in the database.

Parameters:

table (Type[Base]) – Table class (SQLAlchemy declarative base).

Returns:

True if the table exists, False otherwise.

Return type:

bool

class vyra_base.storage.DBTYPE(*values)

Bases: str, Enum

Enumeration of supported database types.

Variables:

• SQLITE – SQLite database engine.

• MYSQL – MySQL database engine.

• POSTGRESQL – PostgreSQL database engine.

SQLITE = 'sqlite'

MYSQL = 'mysql'

POSTGRESQL = 'postgresql'

class vyra_base.storage.DBSTATUS(*values)

Bases: str, Enum

Enumeration of database operation status codes.

Variables:

• SUCCESS – Operation completed successfully.

• ERROR – An error occurred during the operation.

• NOT_FOUND – Requested resource was not found.

• NOT_ALLOWED – Operation is not allowed.

• NOT_AUTHORIZED – User is not authorized for this operation.

• CONFLICT – Operation conflicts with existing data.

SUCCESS = 'success'

ERROR = 'error'

NOT_FOUND = 'not_found'

NOT_ALLOWED = 'not_allowed'

NOT_AUTHORIZED = 'not_authorized'

CONFLICT = 'conflict'

class vyra_base.storage.DBMESSAGE

Bases: object

Collection of standard database error messages.

Variables:

DEFAULT_ERROR – Generic error message for database operations.

DEFAULT_ERROR = 'Something went wrong while processing the query. See the details.'

class vyra_base.storage.DbManipulator(**kwargs)

Bases: object

Datatable class manipulator

__init__(**kwargs)

Wrapper for synchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

get_table_structure()

Read the datatable structure from the config file.

Returns:

Table structure information.

Return type:

DBReturnValue

async get_by_id(id=-1)

Read line from datatable of database by ‘id’.

If the id is -1 or None, the last line will be read.

Parameters:

id (Union[UUID, int]) – Private key of the table element to select the row to be read.

Returns:

Row data or not found status.

Return type:

DBReturnValue

async get_all(filters=None, order_by=None, limit=None)

Read all lines from datatable of database.

Parameters:

• filters (dict) – Filter elements to identify the rows to be read.

• order_by (str, optional) – Column name to order the result.

• limit (int, optional) – Number of lines to be read.

Returns:

List of rows or not found status.

Return type:

DBReturnValue

async update(data, filters=None)

Update a line in a datatable of database by given data.

If the table config has a field ‘max_lines’ and the number of lines are greater than this field, the lowest ‘id’ value will be deleted and the new entry will be added by a incremented ‘id’ value.

Parameters:

• data (dict) – Update data in an existing entry.

• filters (Optional[dict]) – Filter to select the row to be updated.

Returns:

Update status.

Return type:

DBReturnValue

async add(data)

Add a new a row in a datatable.

Parameters:

data (dict) – Content of the new entry to be added to the table.

Returns:

Add status and details.

Return type:

DBReturnValue

async get_one(filters=None)

Read a single row from the database table, optionally filtered.

Returns the first matching model instance using scalars().first(), consistent with get_all(). Use this instead of get_by_id when you need to filter by arbitrary columns or want a guaranteed model instance (not a raw Row).

Parameters:

filters (dict | None) – Optional column/value pairs to filter by.

Returns:

DBReturnValue with .value = model instance or None.

Return type:

DBReturnValue

async add_instance(obj)

Add a pre-built model instance to the database.

Use this instead of add() when you need full control over the object’s fields (e.g. custom primary keys or enum values) and have already constructed the model instance yourself.

Parameters:

obj (Any) – A model instance (subclass of the declarative base).

Returns:

Add status and details.

Return type:

DBReturnValue

async delete(id)

Update a line in a datatable of database by a given ‘id’.

Parameters:

id (Any) – Private key of the table element to select the row to be deleted.

Returns:

Delete status and details.

Return type:

DBReturnValue

async bulk_add(data)

Add multiple rows to a datatable.

Parameters:

data (list[dict]) – Data list to be added to the table.

Returns:

Bulk add status and details.

Return type:

DBReturnValue

async bulk_delete(filters)

Delete multiple rows in a datatable by given filters.

Parameters:

filters (dict) – Filter elements to identify the rows to be deleted.

Returns:

Bulk delete status and details.

Return type:

DBReturnValue

async exists(id)

Check if a row exists in the datatable by a given ‘id’.

Parameters:

id (int) – Private key of the table element to check.

Returns:

Existence status.

Return type:

DBReturnValue

async count(filters)

Count the number of rows in a datatable matching given filters.

Parameters:

filters (dict) – Filter elements to identify the rows to be counted.

Returns:

Number of matching rows.

Return type:

DBReturnValue

to_dict(obj, exclude=None)

Convert a SQLAlchemy model instance to a dictionary.

Parameters:

• obj (Base) – SQLAlchemy model instance.

• exclude (list, optional) – List of column names to exclude.

Returns:

Dictionary representation of the object.

Return type:

dict

class vyra_base.storage.DBReturnValue(status=None, value='', details='')

Bases: object

Standardized return value container for database operations.

Encapsulates the result of database operations with status, value, and details. Provides convenience methods for setting error/success states.

Variables:

• status – Operation status (see DBSTATUS enum).

• value – Main return value (data or error message).

• details – Additional details about the operation.

Parameters:

• status (Union[str, None])

• value (Union[bool, str, list, dict])

• details (Union[bool, str, list, dict])

__init__(status=None, value='', details='')

Parameters:

• status (None | str)

• value (bool | str | list | dict)

• details (bool | str | list | dict)

status: str | None = None

value: bool | str | list | dict = ''

details: bool | str | list | dict = ''

error_return(details='')

Set the return value to error status.

Parameters:

details (Union[bool, str, list, dict]) – Additional error details.

Returns:

Self with error status set.

Return type:

DBReturnValue

success_return()

Set the return value to success status.

Returns:

Self with success status set.

Return type:

DBReturnValue

class vyra_base.storage.RedisClient(**kwargs)

Bases: object

Unified Redis Client for vyra_base Combines RedisAccess and RedisManipulator functionality with TLS support and streaming capabilities for professional communication

__init__(**kwargs)

Wrapper for synchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async connect(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async ping()

Ping Redis server to check connectivity.

Return type:

bool

async configure_base_settings(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async close(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async get(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async set(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async delete(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async exists(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async clear(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async get_all_keys(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async get_keys_by_pattern(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async get_type(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async get_length(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async publish_message(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async subscribe_channel(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async subscribe_to_key(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async unsubscribe_from_key(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async health_check(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async subscribe_pattern(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async publish_with_metadata(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async create_pubsub_listener(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async parse_message(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async get_active_listeners(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async remove_listener_channels(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async stop_pubsub_listener(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async publish_event(channel, data, add_metadata=True)

Alias for publish_with_metadata (backward compatibility)

Return type:

int

Parameters:

• channel (str)

• data (Dict[str, Any])

• add_metadata (bool)

async create_stream_listener(channels, callback_handler, callback_context=None)

Alias for create_pubsub_listener (backward compatibility)

Return type:

None

Parameters:

channels (list[str])

async xadd(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xread(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xreadgroup(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xack(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xgroup_create(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xgroup_destroy(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xlen(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xtrim(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

async xpending(**kwargs)

Wrapper for asynchronous functions.

Parameters:

• args – Positional arguments.

• kwargs – Keyword arguments.

Returns:

Result of the wrapped function.

class vyra_base.storage.REDIS_TYPE(*values)

Bases: str, Enum

Redis data types supported by the client.

STRING = 'string'

HASH = 'hash'

LIST = 'list'

SET = 'set'

class vyra_base.storage.Storage

Bases: object

Base class for storage access.

This class provides a common interface for different storage backends.

It can be used to access different storage backends like databases, files, etc.

vyra_base.storage.tb_parameters

alias of Parameter

vyra_base.storage.tb_error_logs

alias of ErrorLog