Reference

Configuration Matrix

Create Engine Support Functions by Database Type Confirmed by testing [SQLite, PostgreSQL, MSSQL] To Be Tested [MySQL, Oracle] and should be considered experimental.

Option	SQLite	PostgreSQL	MySQL	Oracle	MSSQL
echo	Yes	Yes	Yes	Yes	Yes
future	Yes	Yes	Yes	Yes	Yes
pool_pre_ping	Yes	Yes	Yes	Yes	Yes
pool_size	No	Yes	Yes	Yes	Yes
max_overflow	No	Yes	Yes	Yes	Yes
pool_recycle	Yes	Yes	Yes	Yes	Yes
pool_timeout	No	Yes	Yes	Yes	Yes

In addition to the dialect-specific options above, the following create_async_engine options are accepted regardless of dialect: connect_args, execution_options, isolation_level, query_cache_size, and hide_parameters. Use connect_args for driver-level options such as SSL/TLS settings or connect timeouts.

dsg_lib.async_database_functions.database_operations

This module provides the DatabaseOperations class for performing CRUD operations on a database using SQLAlchemy's asynchronous session.

The DatabaseOperations class includes the following methods:

- `execute_one`: Executes a single non-read SQL query asynchronously.
- `execute_many`: Executes multiple non-read SQL queries asynchronously within a single transaction.
- 'read_one_record': Retrieves a single record from the database based on the provided query.
- `read_query`: Executes a fetch query on the database and returns a list of records that match the query.
- `read_multi_query`: Executes multiple fetch queries on the database and returns a dictionary of results for each query.
- `count_query`: Counts the number of records that match a given query.
- `paginate_query`: Fetches a page of records together with the total matching count.
- `get_column_details`: Gets the details of the columns in a table.
- `get_primary_keys`: Gets the primary keys of a table.
- `get_table_names`: Gets the names of all tables in the database.

Deprecated Methods:
- `create_one`: [Deprecated] Use `execute_one` with an INSERT query instead.
- `create_many`: [Deprecated] Use `execute_many` with INSERT queries instead.
- `update_one`: [Deprecated] Use `execute_one` with an UPDATE query instead.
- `delete_one`: [Deprecated] Use `execute_one` with a DELETE query instead.
- `delete_many`: [Deprecated] Use `execute_many` with DELETE queries instead.

Each method is designed to handle errors correctly and provide a simple interface for performing database operations.

This module also imports the necessary SQLAlchemy and loguru modules, and the AsyncDatabase class from the local async_database module.

Author: Mike Ryan Date: 2024/11/29 License: MIT

DatabaseErrorResult

Bases: dict

A dict subclass returned by DatabaseOperations methods on failure, always of the shape {"error": str, "details": str}.

It behaves exactly like the plain dict callers may already check for (equality, in, .get(), isinstance(result, dict) all work unchanged), so existing code is unaffected. It exists so new code can reliably tell an error result apart from a real one with isinstance(result, DatabaseErrorResult) instead of duck-typing on dict shape, which is ambiguous whenever the real result is itself a dict (e.g. a multi-column row).

Source code in dsg_lib/async_database_functions/database_operations.py

class DatabaseErrorResult(dict):
    """
    A `dict` subclass returned by `DatabaseOperations` methods on failure, always
    of the shape `{"error": str, "details": str}`.

    It behaves exactly like the plain dict callers may already check for
    (equality, `in`, `.get()`, `isinstance(result, dict)` all work unchanged), so
    existing code is unaffected. It exists so new code can reliably tell an error
    result apart from a real one with `isinstance(result, DatabaseErrorResult)`
    instead of duck-typing on dict shape, which is ambiguous whenever the real
    result is itself a dict (e.g. a multi-column row).
    """

DatabaseOperations

This class provides methods for performing CRUD operations on a database using SQLAlchemy's asynchronous session.

The methods include:

• execute_one: Executes a single non-read SQL query asynchronously.
• execute_many: Executes multiple non-read SQL queries asynchronously within a single transaction.
• read_one_record: Retrieves a single record from the database based on the provided query.
• read_query: Executes a fetch query on the database and returns a list of records that match the query.
• read_multi_query: Executes multiple fetch queries on the database and returns a dictionary of results for each query.
• count_query: Counts the number of records that match a given query.
• paginate_query: Fetches a page of records together with the total matching count.
• get_column_details: Gets the details of the columns in a table.
• get_primary_keys: Gets the primary keys of a table.
• get_table_names: Gets the names of all tables in the database.

Deprecated Methods: - create_one: [Deprecated] Use execute_one with an INSERT query instead. - create_many: [Deprecated] Use execute_many with INSERT queries instead. - update_one: [Deprecated] Use execute_one with an UPDATE query instead. - delete_one: [Deprecated] Use execute_one with a DELETE query instead. - delete_many: [Deprecated] Use execute_many with DELETE queries instead.

Examples:

from sqlalchemy import insert, select
from dsg_lib.async_database_functions import (
    async_database,
    base_schema,
    database_config,
    database_operations,
)

# Create a DBConfig instance
config = {
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}
# create database configuration
db_config = database_config.DBConfig(config)
# Create an AsyncDatabase instance
async_db = async_database.AsyncDatabase(db_config)
# Create a DatabaseOperations instance
db_ops = database_operations.DatabaseOperations(async_db)

# create one record
query = insert(User).values(name='John Doe')
result = await db_ops.execute_one(query)

# read one record
query = select(User).where(User.name == 'John Doe')
record = await db_ops.read_query(query)

Source code in dsg_lib/async_database_functions/database_operations.py

class DatabaseOperations:
    """
    This class provides methods for performing CRUD operations on a database using SQLAlchemy's asynchronous session.

    The methods include:

    - `execute_one`: Executes a single non-read SQL query asynchronously.
    - `execute_many`: Executes multiple non-read SQL queries asynchronously within a single transaction.
    - `read_one_record`: Retrieves a single record from the database based on the provided query.
    - `read_query`: Executes a fetch query on the database and returns a list of records that match the query.
    - `read_multi_query`: Executes multiple fetch queries on the database and returns a dictionary of results for each query.
    - `count_query`: Counts the number of records that match a given query.
    - `paginate_query`: Fetches a page of records together with the total matching count.
    - `get_column_details`: Gets the details of the columns in a table.
    - `get_primary_keys`: Gets the primary keys of a table.
    - `get_table_names`: Gets the names of all tables in the database.

    Deprecated Methods:
    - `create_one`: [Deprecated] Use `execute_one` with an INSERT query instead.
    - `create_many`: [Deprecated] Use `execute_many` with INSERT queries instead.
    - `update_one`: [Deprecated] Use `execute_one` with an UPDATE query instead.
    - `delete_one`: [Deprecated] Use `execute_one` with a DELETE query instead.
    - `delete_many`: [Deprecated] Use `execute_many` with DELETE queries instead.

    Examples:
    ```python
    from sqlalchemy import insert, select
    from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
    )

    # Create a DBConfig instance
    config = {
        "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
        "echo": False,
        "future": True,
        "pool_recycle": 3600,
    }
    # create database configuration
    db_config = database_config.DBConfig(config)
    # Create an AsyncDatabase instance
    async_db = async_database.AsyncDatabase(db_config)
    # Create a DatabaseOperations instance
    db_ops = database_operations.DatabaseOperations(async_db)

    # create one record
    query = insert(User).values(name='John Doe')
    result = await db_ops.execute_one(query)

    # read one record
    query = select(User).where(User.name == 'John Doe')
    record = await db_ops.read_query(query)
    ```
    """

    def __init__(self, async_db: AsyncDatabase):
        """
        Initializes a new instance of the DatabaseOperations class.

        Args:
            async_db (AsyncDatabase): An instance of the AsyncDatabase class
            for performing asynchronous database operations.

        See the class docstring above for a complete setup example.
        """
        # Log the start of the initialization
        logger.debug("Initializing DatabaseOperations instance")

        # Store the AsyncDatabase instance in the async_db attribute This
        # instance will be used for performing asynchronous database operations
        self.async_db = async_db

        # Log the successful initialization
        logger.debug("DatabaseOperations instance initialized successfully")

    async def get_columns_details(self, table):
        """
        Retrieves the details of the columns of a given table.

        This asynchronous method accepts a table object and returns a
        dictionary. Each key in the dictionary is a column name from the table,
        and the corresponding value is another dictionary containing details
        about that column, such as type, if it's nullable, if it's a primary
        key, if it's unique, its autoincrement status, and its default value.

        Args:
            table (Table): An instance of the SQLAlchemy Table class
            representing the database table for which column details are
            required.

        Returns:
            dict: A dictionary where each key is a column name, and each value
            is a dictionary with the column's details.

        Raises:
            Exception: If any error occurs during the database operation.

        Example:
            ```python
            columns = await db_ops.get_columns_details(User)
            ```
        """
        # Log the start of the operation
        logger.debug(
            f"Starting get_columns_details operation for table: {table.__name__}"
        )

        try:
            # Log the start of the column retrieval
            logger.debug(f"Getting columns for table: {table.__name__}")

            # Retrieve the details of the columns and store them in a dictionary
            # The keys are the column names and the values are dictionaries
            # containing the column details
            columns = {
                c.name: {
                    "type": str(c.type),
                    "nullable": c.nullable,
                    "primary_key": c.primary_key,
                    "unique": c.unique,
                    "autoincrement": c.autoincrement,
                    "default": (
                        str(c.default.arg)
                        if c.default is not None and not callable(c.default.arg)
                        else None
                    ),
                }
                for c in table.__table__.columns
            }

            # Log the successful column retrieval
            logger.debug(f"Successfully retrieved columns for table: {table.__name__}")

            return columns
        except Exception as ex:  # pragma: no cover
            # Handle any exceptions that occur during the column retrieval
            logger.error(
                f"An error occurred while getting columns for table: {table.__name__}"
            )  # pragma: no cover
            return handle_exceptions(ex)  # pragma: no cover

    async def get_primary_keys(self, table):
        """
        Retrieves the primary keys of a given table.

        This asynchronous method accepts a table object and returns a list
        containing the names of its primary keys. It is useful for understanding
        the structure of the table and for operations that require knowledge of
        the primary keys.

        Args:
            table (Table): An instance of the SQLAlchemy Table class
            representing the database table for which primary keys are required.

        Returns:
            list: A list containing the names of the primary keys of the table.

        Raises:
            Exception: If any error occurs during the database operation.

        Example:
            ```python
            primary_keys = await db_ops.get_primary_keys(User)
            ```
        """
        # Log the start of the operation
        logger.debug(f"Starting get_primary_keys operation for table: {table.__name__}")

        try:
            # Log the start of the primary key retrieval
            logger.debug(f"Getting primary keys for table: {table.__name__}")

            # Retrieve the primary keys and store them in a list
            primary_keys = table.__table__.primary_key.columns.keys()

            # Log the successful primary key retrieval
            logger.debug(f"Primary keys retrieved successfully: {primary_keys}")

            return primary_keys

        except Exception as ex:  # pragma: no cover
            # Handle any exceptions that occur during the primary key retrieval
            logger.error(f"Exception occurred: {ex}")  # pragma: no cover
            return handle_exceptions(ex)  # pragma: no cover

    async def get_table_names(self):
        """
        Retrieves the names of all tables in the database.

        This asynchronous method returns a list containing the names of all
        tables in the database. It is useful for database introspection,
        allowing the user to know which tables are available in the current
        database context.

        Returns:
            list: A list containing the names of all tables in the database.

        Raises:
            Exception: If any error occurs during the database operation.

        Example:
            ```python
            table_names = await db_ops.get_table_names()
            ```
        """
        # Log the start of the operation
        logger.debug("Starting get_table_names operation")

        try:
            # Log the start of the table name retrieval
            logger.debug("Retrieving table names")

            # Retrieve the table names and store them in a list The keys of the
            # metadata.tables dictionary are the table names
            table_names = list(self.async_db.Base.metadata.tables.keys())

            # Log the successful table name retrieval
            logger.debug(f"Table names retrieved successfully: {table_names}")

            return table_names

        except Exception as ex:  # pragma: no cover
            # Handle any exceptions that occur during the table name retrieval
            logger.error(f"Exception occurred: {ex}")  # pragma: no cover
            return handle_exceptions(ex)  # pragma: no cover

    # ---------------------------
    # Internal helpers (refactor)
    # ---------------------------
    def _classify_statement(self, query: ClauseElement) -> str:
        """Classify a SQLAlchemy Core statement into a simple kind.

        Returns one of: 'select' | 'insert' | 'update' | 'delete' | 'other'.
        """
        if _SELECT_TYPE is not None and isinstance(query, _SELECT_TYPE):
            return "select"
        if _INSERT_TYPE is not None and isinstance(query, _INSERT_TYPE):
            return "insert"
        if _UPDATE_TYPE is not None and isinstance(query, _UPDATE_TYPE):
            return "update"
        if _DELETE_TYPE is not None and isinstance(query, _DELETE_TYPE):
            return "delete"
        return "other"

    def _shape_rows_from_result(self, result) -> List[Any]:
        """Shape rows from a SQLAlchemy Result similar to read_query behavior.

        - Single column → list of scalars
        - Multi column → list of dicts via row._mapping
        - ORM/objects → list of objects
        """
        try:
            # Keys available: determine single vs multi column quickly
            if self._has_keys(result):
                if self._is_single_column(result):
                    return result.scalars().all()
                return [self._shape_row(row) for row in result.fetchall()]

            # Fallback when keys() is not available
            return [self._shape_row(row) for row in result.fetchall()]
        except Exception as _row_ex:  # pragma: no cover - defensive
            logger.debug(f"_shape_rows_from_result failed: {_row_ex}")
            return []

    def _has_keys(self, result) -> bool:
        """Return True if result exposes keys() callable."""
        return hasattr(result, "keys") and callable(result.keys)

    def _is_single_column(self, result) -> bool:
        """Return True if result has exactly one column."""
        try:
            keys = result.keys()
            return len(keys) == 1
        except Exception:  # pragma: no cover - defensive
            return False

    def _shape_row(self, row: Any) -> Any:
        """Shape a single row to a scalar/dict/object according to mapping presence."""
        if hasattr(row, "_mapping"):
            mapping = row._mapping
            return list(mapping.values())[0] if len(mapping) == 1 else dict(mapping)
        # For ORM/objects or plain rows, return as-is
        return row

    def _collect_rows(self, result) -> List[Any]:
        """Collect and shape rows from a SQLAlchemy Result if it returns rows."""
        try:
            if getattr(result, "returns_rows", False):
                return self._shape_rows_from_result(result)
        except Exception as _row_ex:  # pragma: no cover - defensive
            logger.debug(f"Unable to materialize returned rows: {_row_ex}")
        return []

    def _assemble_dml_metadata(self, query: ClauseElement, result, rows: List[Any]) -> Dict[str, Any]:
        """Build a metadata dictionary for DML results, including rowcount, inserted PK, rows."""
        meta: Dict[str, Any] = {"rowcount": getattr(result, "rowcount", None)}
        try:
            if _INSERT_TYPE is not None and isinstance(query, _INSERT_TYPE):
                meta["inserted_primary_key"] = getattr(result, "inserted_primary_key", None)
        except Exception:  # pragma: no cover - driver specific
            meta["inserted_primary_key"] = None
        if rows:
            meta["rows"] = rows
        return meta

    async def count_query(self, query):
        """
        Executes a count query on the database and returns the number of records
        that match the query.

        This asynchronous method accepts a SQLAlchemy `Select` query object and
        returns the count of records that match the query. This is particularly
        useful for getting the total number of records that satisfy certain
        conditions without actually fetching the records themselves.

        Parameters:
            query (Select): A SQLAlchemy `Select` query object specifying the
            conditions to count records for.

        Returns:
            int: The number of records that match the query.

        Raises:
            Exception: If any error occurs during the execution of the query.

        Example:
            ```python
            count = await db_ops.count_query(select(User).where(User.age > 30))
            ```
        """
        # Log the start of the operation
        logger.debug("Starting count_query operation")

        try:
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the query being executed
                logger.debug(f"Executing count query: {query}")

                # Execute the count query and retrieve the count
                result = await session.execute(
                    select(func.count()).select_from(query.subquery())
                )
                count = result.scalar()

                # Log the successful query execution
                logger.debug(f"Count query executed successfully. Result: {count}")

                return count

        except Exception as ex:
            # Handle any exceptions that occur during the query execution
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    async def paginate_query(
        self,
        query,
        page: int = 1,
        page_size: int = 100,
    ) -> Union[Dict[str, Any], DatabaseErrorResult]:
        """
        Executes a query with page-based pagination, returning the requested
        page of records together with the total matching record count.

        This combines what would otherwise be a separate `count_query` call and
        a `read_query` call (each opening their own session) into one method
        that counts and fetches the page within a single session.

        Args:
            query (Select): A SQLAlchemy `Select` query object (without its own
                `.limit()`/`.offset()` applied) specifying the records to page
                through. Apply any `.where()`/`.order_by()` clauses beforehand.
            page (int): The 1-indexed page number to fetch. Defaults to 1.
            page_size (int): The number of records per page. Defaults to 100.

        Returns:
            dict: `{"items": List[Any], "total": int, "page": int,
            "page_size": int, "pages": int}`, where `items` is the shaped page
            of records (scalars, dicts, or ORM objects, as in `read_query`) and
            `pages` is the total number of pages (0 if there are no matches).
            DatabaseErrorResult: If `page`/`page_size` are invalid, or an error
            occurs during the query.

        Example:
            ```python
            result = await db_ops.paginate_query(select(User).order_by(User.name), page=2, page_size=25)
            # result["items"], result["total"], result["pages"]
            ```
        """
        if page < 1 or page_size < 1:
            return DatabaseErrorResult(
                {
                    "error": "Invalid pagination parameters",
                    "details": f"page and page_size must be >= 1 (got page={page}, page_size={page_size})",
                }
            )

        logger.debug(f"Starting paginate_query operation: page={page}, page_size={page_size}")

        try:
            async with self.async_db.get_db_session() as session:
                count_result = await session.execute(
                    select(func.count()).select_from(query.subquery())
                )
                total = count_result.scalar()

                paged_query = query.limit(page_size).offset((page - 1) * page_size)
                result = await session.execute(paged_query)
                items = self._shape_rows_from_result(result)

                pages = (total + page_size - 1) // page_size if total else 0

                logger.debug(f"paginate_query result: total={total}, pages={pages}")
                return {
                    "items": items,
                    "total": total,
                    "page": page,
                    "page_size": page_size,
                    "pages": pages,
                }

        except Exception as ex:
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    async def read_one_record(self, query):
        """
        Retrieves a single record from the database based on the provided query.

        This asynchronous method accepts a SQL query object and returns the
        first record that matches the query. If no record matches the query, it
        returns None. This method is useful for fetching specific data
        when the expected result is a single record.

        Parameters:
            query (Select): An instance of the SQLAlchemy Select class,
            representing the query to be executed.

        Returns:
            Result: The first record that matches the query or None if no record matches.

        Raises:
            Exception: If any error occurs during the database operation.

        Example:
            ```python
            record = await db_ops.read_one_record(select(User).where(User.name == 'John Doe'))
            ```
        """
        # Log the start of the operation
        logger.debug(f"Starting read_one_record operation for {query}")

        try:
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the start of the record retrieval
                logger.debug(f"Getting record with query: {query}")

                # Execute the query and retrieve the first record
                result = await session.execute(query)
                record = result.scalar_one()

                # Log the successful record retrieval
                logger.debug(f"Record retrieved successfully: {record}")

                return record

        except NoResultFound:
            # No record was found
            logger.debug("No record found")
            return None

        except Exception as ex:  # pragma: no cover
            # Handle any exceptions that occur during the record retrieval
            logger.error(f"Exception occurred: {ex}")  # pragma: no cover
            return handle_exceptions(ex)  # pragma: no cover

    async def read_query(self, query):
        # Log the start of the operation
        logger.debug("Starting read_query operation")

        try:
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the query being executed
                logger.debug(f"Executing fetch query: {query}")

                # Execute the fetch query and retrieve the records
                result = await session.execute(query)
                records = self._shape_rows_from_result(result)

                logger.debug(f"read_query result: {records}")
                return records

        except Exception as ex:
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    async def read_multi_query(self, queries: Dict[str, str]):
        """
        Executes multiple fetch queries asynchronously and returns a dictionary of results for each query.

        This asynchronous method accepts a dictionary where each key is a query name (str)
        and each value is a SQLAlchemy `Select` query object. It executes each query within a single
        database session and collects the results. The results are returned as a dictionary mapping
        each query name to a list of records that match that query.

        The function automatically determines the structure of each result set:
        - If the query returns a single column, the result will be a list of scalar values.
        - If the query returns multiple columns, the result will be a list of dictionaries mapping column names to values.
        - If the result row is an ORM object, it will be returned as-is.

        Args:
            queries (Dict[str, Select]): A dictionary mapping query names to SQLAlchemy `Select` query objects.

        Returns:
            Dict[str, List[Any]]: A dictionary where each key is a query name and each value is a list of records
            (scalars, dictionaries, or ORM objects) that match the corresponding query.

        Raises:
            Exception: If any error occurs during the execution of any query, the function logs the error and
            returns a dictionary with error details using `handle_exceptions`.

        Example:
            ```python
            from sqlalchemy import select
            queries = {
                "adults": select(User).where(User.age >= 18),
                "minors": select(User).where(User.age < 18),
            }
            results = await db_ops.read_multi_query(queries)
            # results["adults"] and results["minors"] will contain lists of records
            ```
        """
        # Log the start of the operation
        logger.debug("Starting read_multi_query operation")

        try:
            results = {}
            async with self.async_db.get_db_session() as session:
                for query_name, query in queries.items():
                    logger.debug(f"Executing fetch query: {query}")
                    result = await session.execute(query)
                    results[query_name] = self._shape_rows_from_result(result)
            return results

        except Exception as ex:
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    async def execute_one(
        self,
        query: ClauseElement,
        values: Optional[Dict[str, Any]] = None,
        return_metadata: bool = False,
    ) -> Union[str, Dict[str, Any]]:
        """
                Executes a single SQL statement asynchronously and returns a meaningful result.

                Supports both read (SELECT) and write (INSERT/UPDATE/DELETE) statements.
                - For SELECT, returns the fetched records (list of scalars, dicts, or ORM objects).
                - For INSERT/UPDATE/DELETE, returns a dict with metadata including rowcount,
                    inserted_primary_key (if available), and any returned rows when the statement
                    includes RETURNING.

        Args:
            query (ClauseElement): An SQLAlchemy query object representing the SQL statement to execute.
            values (Optional[Dict[str, Any]]): A dictionary of parameter values to bind to the query.
                Defaults to None.
            return_metadata (bool): When True, returns metadata for DML statements.
                When False (default), returns "complete" for DML statements to preserve legacy behavior.

                Returns:
                        - For SELECT: List[Any] of records.
                        - For DML (INSERT/UPDATE/DELETE) when return_metadata=True: Dict with keys:
                                {
                                    "rowcount": int | None,
                                    "inserted_primary_key": List[Any] | None,
                                    "rows": List[Any] (if statement returns rows)
                                }
                        - For DML when return_metadata=False: the string "complete".
                        - On error: Dict[str, str] from handle_exceptions.

        Example:
            ```python
            from sqlalchemy import insert

            query = insert(User).values(name='John Doe')
            result = await db_ops.execute_one(query)
            ```
        """
        logger.debug("Starting execute_one operation")
        try:
            async with self.async_db.get_db_session() as session:
                logger.debug(f"Executing query: {query}")
                result = await session.execute(query, params=values)

                # SELECT → shape and return records directly from this same
                # session/transaction; no commit needed for a pure read.
                if self._classify_statement(query) == "select":
                    records = self._shape_rows_from_result(result)
                    logger.debug(f"execute_one result: {records}")
                    return records

                rows = self._collect_rows(result)
                meta = self._assemble_dml_metadata(query, result, rows)

                # Commit for non-SELECT statements (DML/DDL).
                await session.commit()

                logger.debug(f"Query executed successfully with metadata: {meta}")
                return meta if return_metadata else "complete"
        except Exception as ex:
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    async def execute_many(
        self,
        queries: List[Tuple[ClauseElement, Optional[Dict[str, Any]]]],
        return_results: bool = False,
    ) -> Union[str, List[Any], Dict[str, str]]:
        """
                Executes multiple SQL statements asynchronously within a single transaction.

                        Supports a mix of SELECT and DML statements.
                        When return_results=True, returns a list of per-statement results in the order provided.
                        Each item is either:
                        - For SELECT: a list of records.
                        - For DML: a dict containing rowcount, inserted_primary_key (if available),
                            and any returned rows (when using RETURNING).

        Args:
            queries (List[Tuple[ClauseElement, Optional[Dict[str, Any]]]]): A list of tuples, each containing
                a query and an optional dictionary of parameter values. Each tuple should be of the form
                `(query, values)` where:
                    - `query` is an SQLAlchemy query object.
                    - `values` is a dictionary of parameters to bind to the query (or None).

        Returns:
            - On success and return_results=True: List[Any | Dict[str, Any]] with one entry per input query.
            - On success and return_results=False: the string "complete" (legacy behavior).
            - On error: Dict[str, str] from handle_exceptions.

        Example:
            ```python
            from sqlalchemy import insert

            queries = [
                (insert(User), {'name': 'User1'}),
                (insert(User), {'name': 'User2'}),
                (insert(User), {'name': 'User3'}),
            ]
            result = await db_ops.execute_many(queries)
            ```
        """
        logger.debug("Starting execute_many operation")
        try:
            results: List[Any] = []
            async with self.async_db.get_db_session() as session:
                for query, values in queries:
                    logger.debug(f"Executing query: {query}")
                    res = await session.execute(query, params=values)

                    # SELECT → shape rows from this same session/transaction,
                    # so it sees writes made earlier in this same batch.
                    if self._classify_statement(query) == "select":
                        data = self._shape_rows_from_result(res)
                        if return_results:
                            results.append(data)
                        continue

                    rows = self._collect_rows(res)
                    meta = self._assemble_dml_metadata(query, res, rows)
                    if return_results:
                        results.append(meta)

                # Commit once for the whole batch (safe even if some were SELECT)
                await session.commit()

            logger.debug("All queries executed successfully with results collected")
            return results if return_results else "complete"
        except Exception as ex:
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    @deprecated("Use `execute_one` with an INSERT query instead.")
    async def create_one(self, record):
        """
        This method is deprecated. Use `execute_one` with an INSERT query instead.

        Adds a single record to the database.

        This asynchronous method accepts a record object and adds it to the
        database. If the operation is successful, it returns the added record.
        The method is useful for inserting a new row into a database table.

        Parameters:
            record (Base): An instance of the SQLAlchemy declarative base class
            representing the record to be added to the database.

        Returns:
            Base: The instance of the record that was added to the database.

        Raises:
            Exception: If any error occurs during the database operation.

        Example:
            ```python
            record = await db_ops.create_one(User(name='John Doe'))
            ```
        """
        # Log the start of the operation
        logger.debug("Starting create_one operation")

        try:
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the record being added
                logger.debug(f"Adding record to session: {record.__dict__}")

                # Add the record to the session and commit the changes
                session.add(record)
                await session.commit()

                # Log the successful record addition
                logger.debug(f"Record added successfully: {record}")

                return record

        except Exception as ex:
            # Handle any exceptions that occur during the record addition
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    @deprecated("Use `execute_one` with an INSERT query instead.")
    async def create_many(self, records):
        """
        This method is deprecated. Use `execute_many` with INSERT queries instead.

        Adds multiple records to the database.

        This asynchronous method accepts a list of record objects and adds them
        to the database. If the operation is successful, it returns the added
        records. This method is useful for bulk inserting multiple rows into a
        database table efficiently.

        Parameters:
            records (list[Base]): A list of instances of the SQLAlchemy
            declarative base class, each representing a record to be added to
            the database.

        Returns:
            list[Base]: A list of instances of the records that were added to
            the database.

        Raises:
            Exception: If any error occurs during the database operation.

        Example:
            ```python
            records = await db_ops.create_many([User(name='John Doe'), User(name='Jane Doe')])
            ```
        """
        # Log the start of the operation
        logger.debug("Starting create_many operation")

        try:
            # Start a timer to measure the operation time
            t0 = time.time()

            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the number of records being added
                logger.debug(f"Adding {len(records)} records to session")

                # Add the records to the session and commit the changes
                session.add_all(records)
                await session.commit()

                # Log the added records
                records_data = [record.__dict__ for record in records]
                logger.debug(f"Records added to session: {records_data}")

                # Calculate the operation time and log the successful record
                # addition
                num_records = len(records)
                t1 = time.time() - t0
                logger.debug(
                    f"Record operations were successful. {num_records} records were created in {t1:.4f} seconds."
                )

                return records

        except Exception as ex:
            # Handle any exceptions that occur during the record addition
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    @deprecated("Use `execute_one` with a UPDATE query instead.")
    async def update_one(self, table, record_id: str, new_values: dict):
        """
        This method is deprecated. Use `execute_one` with an UPDATE query instead.

        Updates a single record in the database identified by its ID.

        This asynchronous method takes a SQLAlchemy `Table` object, a record ID,
        and a dictionary of new values to update the record. It updates the
        specified record in the given table with the new values. The method does
        not allow updating certain fields, such as 'id' or 'date_created'.

        Parameters:
            table (Table): The SQLAlchemy `Table` object representing the table
            in the database. record_id (str): The ID of the record to be
            updated. new_values (dict): A dictionary containing the fields to
            update and their new values.

        Returns:
            Base: The updated record if successful; otherwise, an error
            dictionary.

        Raises:
            Exception: If any error occurs during the update operation.

        Example:
            ```python
            record = await db_ops.update_one(User, 1, {'name': 'John Smith'})
            ```
        """
        non_updatable_fields = ["id", "date_created"]

        # Log the start of the operation
        logger.debug(
            f"Starting update_one operation for record_id: {record_id} in table: {table.__name__}"
        )

        try:
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the record being fetched
                logger.debug(f"Fetching record with id: {record_id}")

                # Fetch the record
                record = await session.get(table, record_id)
                if not record:
                    # Log the error if no record is found
                    logger.error(f"No record found with pkid: {record_id}")
                    return DatabaseErrorResult(
                        {
                            "error": "Record not found",
                            "details": f"No record found with pkid {record_id}",
                        }
                    )

                # Log the record being updated
                logger.debug(f"Updating record with new values: {new_values}")

                # Update the record with the new values
                for key, value in new_values.items():
                    if key not in non_updatable_fields:
                        setattr(record, key, value)
                await session.commit()

                # Log the successful record update
                logger.debug(f"Record updated successfully: {record.pkid}")
                return record

        except Exception as ex:
            # Handle any exceptions that occur during the record update
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    @deprecated("Use `execute_many` with a DELETE query instead.")
    async def delete_one(self, table, record_id: str):
        """
        This method is deprecated. Use `execute_one` with a DELETE query instead.

        Deletes a single record from the database based on the provided table
        and record ID.

        This asynchronous method accepts a SQLAlchemy `Table` object and a
        record ID. It attempts to delete the record with the given ID from the
        specified table. If the record is successfully deleted, it returns a
        success message. If no record with the given ID is found, it returns an
        error message.

        Args:
            table (Table): An instance of the SQLAlchemy `Table` class
            representing the database table from which the record will be
            deleted. record_id (str): The ID of the record to be deleted.

        Returns:
            dict: A dictionary containing a success message if the record was
            deleted successfully, or an error message if the record was not
            found or an exception occurred.

        Raises:
            Exception: If any error occurs during the delete operation.

        Example:
            ```python
            result = await db_ops.delete_one(User, 1)
            ```
        """
        # Log the start of the operation
        logger.debug(
            f"Starting delete_one operation for record_id: {record_id} in table: {table.__name__}"
        )

        try:
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the record being fetched
                logger.debug(f"Fetching record with id: {record_id}")

                # Fetch the record
                record = await session.get(table, record_id)

                # If the record doesn't exist, return an error
                if not record:
                    logger.error(f"No record found with pkid: {record_id}")
                    return DatabaseErrorResult(
                        {
                            "error": "Record not found",
                            "details": f"No record found with pkid {record_id}",
                        }
                    )

                # Log the record being deleted
                logger.debug(f"Deleting record with id: {record_id}")

                # Delete the record
                await session.delete(record)

                # Log the successful record deletion from the session
                logger.debug(f"Record deleted from session: {record}")

                # Log the start of the commit
                logger.debug(
                    f"Committing changes to delete record with id: {record_id}"
                )

                # Commit the changes
                await session.commit()

                # Log the successful record deletion
                logger.debug(f"Record deleted successfully: {record_id}")

                return {"success": "Record deleted successfully"}

        except Exception as ex:
            # Handle any exceptions that occur during the record deletion
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

    @deprecated("User 'execute_many' with a DELETE query instead.")
    async def delete_many(
        self,
        table: Type[DeclarativeMeta],
        id_column_name: str = "pkid",
        id_values: List[int] = None,
    ) -> int:
        """
        This method is deprecated. Use `execute_many` with a DELETE query instead.

        Deletes multiple records from the specified table in the database.

        This method takes a table, an optional id column name, and a list of id values. It deletes the records in the table where the id column matches any of the id values in the list.

        Args:
            table (Type[DeclarativeMeta]): The table from which to delete records.
            id_column_name (str, optional): The name of the id column in the table. Defaults to "pkid".
            id_values (List[int], optional): A list of id values for the records to delete. Defaults to [].

        Returns:
            int: The number of records deleted from the table.

        Example:
            ```python
            deleted_count = await db_ops.delete_many(User, 'id', [1, 2, 3])
            ```
        """
        if id_values is None:  # pragma: no cover
            id_values = []
        try:
            # Start a timer to measure the operation time
            t0 = time.time()

            # Start a new database session
            async with self.async_db.get_db_session() as session:
                # Log the number of records being deleted
                logger.debug(f"Deleting {len(id_values)} records from session")

                # Create delete statement
                stmt = delete(table).where(
                    getattr(table, id_column_name).in_(id_values)
                )

                # Execute the delete statement and fetch result
                result = await session.execute(stmt)

                # Commit the changes
                await session.commit()

                # Get the count of deleted records
                deleted_count = result.rowcount

                # Log the deleted records
                logger.debug(f"Records deleted from session: {deleted_count}")

                # Calculate the operation time and log the successful record deletion
                t1 = time.time() - t0
                logger.debug(
                    f"Record operations were successful. {deleted_count} records were deleted in {t1:.4f} seconds."
                )

                return deleted_count

        except Exception as ex:
            # Handle any exceptions that occur during the record deletion
            logger.error(f"Exception occurred: {ex}")
            return handle_exceptions(ex)

__init__(async_db)

Initializes a new instance of the DatabaseOperations class.

Parameters:

Name	Type	Description	Default
async_db	AsyncDatabase	An instance of the AsyncDatabase class	required

See the class docstring above for a complete setup example.

Source code in dsg_lib/async_database_functions/database_operations.py

def __init__(self, async_db: AsyncDatabase):
    """
    Initializes a new instance of the DatabaseOperations class.

    Args:
        async_db (AsyncDatabase): An instance of the AsyncDatabase class
        for performing asynchronous database operations.

    See the class docstring above for a complete setup example.
    """
    # Log the start of the initialization
    logger.debug("Initializing DatabaseOperations instance")

    # Store the AsyncDatabase instance in the async_db attribute This
    # instance will be used for performing asynchronous database operations
    self.async_db = async_db

    # Log the successful initialization
    logger.debug("DatabaseOperations instance initialized successfully")

count_query(query) async

Executes a count query on the database and returns the number of records that match the query.

This asynchronous method accepts a SQLAlchemy Select query object and returns the count of records that match the query. This is particularly useful for getting the total number of records that satisfy certain conditions without actually fetching the records themselves.

Parameters:

Name	Type	Description	Default
query	Select	A SQLAlchemy Select query object specifying the	required

Returns:

Name	Type	Description
int		The number of records that match the query.

Raises:

Type	Description
Exception	If any error occurs during the execution of the query.

Example

count = await db_ops.count_query(select(User).where(User.age > 30))

Source code in dsg_lib/async_database_functions/database_operations.py

async def count_query(self, query):
    """
    Executes a count query on the database and returns the number of records
    that match the query.

    This asynchronous method accepts a SQLAlchemy `Select` query object and
    returns the count of records that match the query. This is particularly
    useful for getting the total number of records that satisfy certain
    conditions without actually fetching the records themselves.

    Parameters:
        query (Select): A SQLAlchemy `Select` query object specifying the
        conditions to count records for.

    Returns:
        int: The number of records that match the query.

    Raises:
        Exception: If any error occurs during the execution of the query.

    Example:
        ```python
        count = await db_ops.count_query(select(User).where(User.age > 30))
        ```
    """
    # Log the start of the operation
    logger.debug("Starting count_query operation")

    try:
        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the query being executed
            logger.debug(f"Executing count query: {query}")

            # Execute the count query and retrieve the count
            result = await session.execute(
                select(func.count()).select_from(query.subquery())
            )
            count = result.scalar()

            # Log the successful query execution
            logger.debug(f"Count query executed successfully. Result: {count}")

            return count

    except Exception as ex:
        # Handle any exceptions that occur during the query execution
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

create_many(records) async

This method is deprecated. Use execute_many with INSERT queries instead.

Adds multiple records to the database.

This asynchronous method accepts a list of record objects and adds them to the database. If the operation is successful, it returns the added records. This method is useful for bulk inserting multiple rows into a database table efficiently.

Parameters:

Name	Type	Description	Default
records	list[Base]	A list of instances of the SQLAlchemy	required

Returns:

Type	Description
	list[Base]: A list of instances of the records that were added to
	the database.

Raises:

Type	Description
Exception	If any error occurs during the database operation.

Example

records = await db_ops.create_many([User(name='John Doe'), User(name='Jane Doe')])

Source code in dsg_lib/async_database_functions/database_operations.py

@deprecated("Use `execute_one` with an INSERT query instead.")
async def create_many(self, records):
    """
    This method is deprecated. Use `execute_many` with INSERT queries instead.

    Adds multiple records to the database.

    This asynchronous method accepts a list of record objects and adds them
    to the database. If the operation is successful, it returns the added
    records. This method is useful for bulk inserting multiple rows into a
    database table efficiently.

    Parameters:
        records (list[Base]): A list of instances of the SQLAlchemy
        declarative base class, each representing a record to be added to
        the database.

    Returns:
        list[Base]: A list of instances of the records that were added to
        the database.

    Raises:
        Exception: If any error occurs during the database operation.

    Example:
        ```python
        records = await db_ops.create_many([User(name='John Doe'), User(name='Jane Doe')])
        ```
    """
    # Log the start of the operation
    logger.debug("Starting create_many operation")

    try:
        # Start a timer to measure the operation time
        t0 = time.time()

        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the number of records being added
            logger.debug(f"Adding {len(records)} records to session")

            # Add the records to the session and commit the changes
            session.add_all(records)
            await session.commit()

            # Log the added records
            records_data = [record.__dict__ for record in records]
            logger.debug(f"Records added to session: {records_data}")

            # Calculate the operation time and log the successful record
            # addition
            num_records = len(records)
            t1 = time.time() - t0
            logger.debug(
                f"Record operations were successful. {num_records} records were created in {t1:.4f} seconds."
            )

            return records

    except Exception as ex:
        # Handle any exceptions that occur during the record addition
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

create_one(record) async

This method is deprecated. Use execute_one with an INSERT query instead.

Adds a single record to the database.

This asynchronous method accepts a record object and adds it to the database. If the operation is successful, it returns the added record. The method is useful for inserting a new row into a database table.

Parameters:

Name	Type	Description	Default
record	Base	An instance of the SQLAlchemy declarative base class	required

Returns:

Name	Type	Description
Base		The instance of the record that was added to the database.

Raises:

Type	Description
Exception	If any error occurs during the database operation.

Example

record = await db_ops.create_one(User(name='John Doe'))

Source code in dsg_lib/async_database_functions/database_operations.py

@deprecated("Use `execute_one` with an INSERT query instead.")
async def create_one(self, record):
    """
    This method is deprecated. Use `execute_one` with an INSERT query instead.

    Adds a single record to the database.

    This asynchronous method accepts a record object and adds it to the
    database. If the operation is successful, it returns the added record.
    The method is useful for inserting a new row into a database table.

    Parameters:
        record (Base): An instance of the SQLAlchemy declarative base class
        representing the record to be added to the database.

    Returns:
        Base: The instance of the record that was added to the database.

    Raises:
        Exception: If any error occurs during the database operation.

    Example:
        ```python
        record = await db_ops.create_one(User(name='John Doe'))
        ```
    """
    # Log the start of the operation
    logger.debug("Starting create_one operation")

    try:
        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the record being added
            logger.debug(f"Adding record to session: {record.__dict__}")

            # Add the record to the session and commit the changes
            session.add(record)
            await session.commit()

            # Log the successful record addition
            logger.debug(f"Record added successfully: {record}")

            return record

    except Exception as ex:
        # Handle any exceptions that occur during the record addition
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

delete_many(table, id_column_name='pkid', id_values=None) async

This method is deprecated. Use execute_many with a DELETE query instead.

Deletes multiple records from the specified table in the database.

This method takes a table, an optional id column name, and a list of id values. It deletes the records in the table where the id column matches any of the id values in the list.

Parameters:

Name	Type	Description	Default
table	Type[DeclarativeMeta]	The table from which to delete records.	required
id_column_name	str	The name of the id column in the table. Defaults to "pkid".	'pkid'
id_values	List[int]	A list of id values for the records to delete. Defaults to [].	None

Returns:

Name	Type	Description
int	int	The number of records deleted from the table.

Example

deleted_count = await db_ops.delete_many(User, 'id', [1, 2, 3])

Source code in dsg_lib/async_database_functions/database_operations.py

@deprecated("User 'execute_many' with a DELETE query instead.")
async def delete_many(
    self,
    table: Type[DeclarativeMeta],
    id_column_name: str = "pkid",
    id_values: List[int] = None,
) -> int:
    """
    This method is deprecated. Use `execute_many` with a DELETE query instead.

    Deletes multiple records from the specified table in the database.

    This method takes a table, an optional id column name, and a list of id values. It deletes the records in the table where the id column matches any of the id values in the list.

    Args:
        table (Type[DeclarativeMeta]): The table from which to delete records.
        id_column_name (str, optional): The name of the id column in the table. Defaults to "pkid".
        id_values (List[int], optional): A list of id values for the records to delete. Defaults to [].

    Returns:
        int: The number of records deleted from the table.

    Example:
        ```python
        deleted_count = await db_ops.delete_many(User, 'id', [1, 2, 3])
        ```
    """
    if id_values is None:  # pragma: no cover
        id_values = []
    try:
        # Start a timer to measure the operation time
        t0 = time.time()

        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the number of records being deleted
            logger.debug(f"Deleting {len(id_values)} records from session")

            # Create delete statement
            stmt = delete(table).where(
                getattr(table, id_column_name).in_(id_values)
            )

            # Execute the delete statement and fetch result
            result = await session.execute(stmt)

            # Commit the changes
            await session.commit()

            # Get the count of deleted records
            deleted_count = result.rowcount

            # Log the deleted records
            logger.debug(f"Records deleted from session: {deleted_count}")

            # Calculate the operation time and log the successful record deletion
            t1 = time.time() - t0
            logger.debug(
                f"Record operations were successful. {deleted_count} records were deleted in {t1:.4f} seconds."
            )

            return deleted_count

    except Exception as ex:
        # Handle any exceptions that occur during the record deletion
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

delete_one(table, record_id) async

This method is deprecated. Use execute_one with a DELETE query instead.

Deletes a single record from the database based on the provided table and record ID.

This asynchronous method accepts a SQLAlchemy Table object and a record ID. It attempts to delete the record with the given ID from the specified table. If the record is successfully deleted, it returns a success message. If no record with the given ID is found, it returns an error message.

Parameters:

Name	Type	Description	Default
table	Table	An instance of the SQLAlchemy Table class	required
deleted. record_id	str	The ID of the record to be deleted.	required

Returns:

Name	Type	Description
dict		A dictionary containing a success message if the record was
		deleted successfully, or an error message if the record was not
		found or an exception occurred.

Raises:

Type	Description
Exception	If any error occurs during the delete operation.

Example

result = await db_ops.delete_one(User, 1)

Source code in dsg_lib/async_database_functions/database_operations.py

@deprecated("Use `execute_many` with a DELETE query instead.")
async def delete_one(self, table, record_id: str):
    """
    This method is deprecated. Use `execute_one` with a DELETE query instead.

    Deletes a single record from the database based on the provided table
    and record ID.

    This asynchronous method accepts a SQLAlchemy `Table` object and a
    record ID. It attempts to delete the record with the given ID from the
    specified table. If the record is successfully deleted, it returns a
    success message. If no record with the given ID is found, it returns an
    error message.

    Args:
        table (Table): An instance of the SQLAlchemy `Table` class
        representing the database table from which the record will be
        deleted. record_id (str): The ID of the record to be deleted.

    Returns:
        dict: A dictionary containing a success message if the record was
        deleted successfully, or an error message if the record was not
        found or an exception occurred.

    Raises:
        Exception: If any error occurs during the delete operation.

    Example:
        ```python
        result = await db_ops.delete_one(User, 1)
        ```
    """
    # Log the start of the operation
    logger.debug(
        f"Starting delete_one operation for record_id: {record_id} in table: {table.__name__}"
    )

    try:
        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the record being fetched
            logger.debug(f"Fetching record with id: {record_id}")

            # Fetch the record
            record = await session.get(table, record_id)

            # If the record doesn't exist, return an error
            if not record:
                logger.error(f"No record found with pkid: {record_id}")
                return DatabaseErrorResult(
                    {
                        "error": "Record not found",
                        "details": f"No record found with pkid {record_id}",
                    }
                )

            # Log the record being deleted
            logger.debug(f"Deleting record with id: {record_id}")

            # Delete the record
            await session.delete(record)

            # Log the successful record deletion from the session
            logger.debug(f"Record deleted from session: {record}")

            # Log the start of the commit
            logger.debug(
                f"Committing changes to delete record with id: {record_id}"
            )

            # Commit the changes
            await session.commit()

            # Log the successful record deletion
            logger.debug(f"Record deleted successfully: {record_id}")

            return {"success": "Record deleted successfully"}

    except Exception as ex:
        # Handle any exceptions that occur during the record deletion
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

execute_many(queries, return_results=False) async

    Executes multiple SQL statements asynchronously within a single transaction.

            Supports a mix of SELECT and DML statements.
            When return_results=True, returns a list of per-statement results in the order provided.
            Each item is either:
            - For SELECT: a list of records.
            - For DML: a dict containing rowcount, inserted_primary_key (if available),
                and any returned rows (when using RETURNING).

Parameters:

Name	Type	Description	Default
queries	List[Tuple[ClauseElement, Optional[Dict[str, Any]]]]	A list of tuples, each containing a query and an optional dictionary of parameter values. Each tuple should be of the form (query, values) where: - query is an SQLAlchemy query object. - values is a dictionary of parameters to bind to the query (or None).	required

Returns:

Type	Description
Union[str, List[Any], Dict[str, str]]	On success and return_results=True: List[Any | Dict[str, Any]] with one entry per input query.
Union[str, List[Any], Dict[str, str]]	On success and return_results=False: the string "complete" (legacy behavior).
Union[str, List[Any], Dict[str, str]]	On error: Dict[str, str] from handle_exceptions.

Example

from sqlalchemy import insert

queries = [
    (insert(User), {'name': 'User1'}),
    (insert(User), {'name': 'User2'}),
    (insert(User), {'name': 'User3'}),
]
result = await db_ops.execute_many(queries)

Source code in dsg_lib/async_database_functions/database_operations.py

async def execute_many(
    self,
    queries: List[Tuple[ClauseElement, Optional[Dict[str, Any]]]],
    return_results: bool = False,
) -> Union[str, List[Any], Dict[str, str]]:
    """
            Executes multiple SQL statements asynchronously within a single transaction.

                    Supports a mix of SELECT and DML statements.
                    When return_results=True, returns a list of per-statement results in the order provided.
                    Each item is either:
                    - For SELECT: a list of records.
                    - For DML: a dict containing rowcount, inserted_primary_key (if available),
                        and any returned rows (when using RETURNING).

    Args:
        queries (List[Tuple[ClauseElement, Optional[Dict[str, Any]]]]): A list of tuples, each containing
            a query and an optional dictionary of parameter values. Each tuple should be of the form
            `(query, values)` where:
                - `query` is an SQLAlchemy query object.
                - `values` is a dictionary of parameters to bind to the query (or None).

    Returns:
        - On success and return_results=True: List[Any | Dict[str, Any]] with one entry per input query.
        - On success and return_results=False: the string "complete" (legacy behavior).
        - On error: Dict[str, str] from handle_exceptions.

    Example:
        ```python
        from sqlalchemy import insert

        queries = [
            (insert(User), {'name': 'User1'}),
            (insert(User), {'name': 'User2'}),
            (insert(User), {'name': 'User3'}),
        ]
        result = await db_ops.execute_many(queries)
        ```
    """
    logger.debug("Starting execute_many operation")
    try:
        results: List[Any] = []
        async with self.async_db.get_db_session() as session:
            for query, values in queries:
                logger.debug(f"Executing query: {query}")
                res = await session.execute(query, params=values)

                # SELECT → shape rows from this same session/transaction,
                # so it sees writes made earlier in this same batch.
                if self._classify_statement(query) == "select":
                    data = self._shape_rows_from_result(res)
                    if return_results:
                        results.append(data)
                    continue

                rows = self._collect_rows(res)
                meta = self._assemble_dml_metadata(query, res, rows)
                if return_results:
                    results.append(meta)

            # Commit once for the whole batch (safe even if some were SELECT)
            await session.commit()

        logger.debug("All queries executed successfully with results collected")
        return results if return_results else "complete"
    except Exception as ex:
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

execute_one(query, values=None, return_metadata=False) async

    Executes a single SQL statement asynchronously and returns a meaningful result.

    Supports both read (SELECT) and write (INSERT/UPDATE/DELETE) statements.
    - For SELECT, returns the fetched records (list of scalars, dicts, or ORM objects).
    - For INSERT/UPDATE/DELETE, returns a dict with metadata including rowcount,
        inserted_primary_key (if available), and any returned rows when the statement
        includes RETURNING.

Parameters:

Name	Type	Description	Default
query	ClauseElement	An SQLAlchemy query object representing the SQL statement to execute.	required
values	Optional[Dict[str, Any]]	A dictionary of parameter values to bind to the query. Defaults to None.	None
return_metadata	bool	When True, returns metadata for DML statements. When False (default), returns "complete" for DML statements to preserve legacy behavior. Returns: - For SELECT: List[Any] of records. - For DML (INSERT/UPDATE/DELETE) when return_metadata=True: Dict with keys: { "rowcount": int | None, "inserted_primary_key": List[Any] | None, "rows": List[Any] (if statement returns rows) } - For DML when return_metadata=False: the string "complete". - On error: Dict[str, str] from handle_exceptions.	False

Example

from sqlalchemy import insert

query = insert(User).values(name='John Doe')
result = await db_ops.execute_one(query)

Source code in dsg_lib/async_database_functions/database_operations.py

async def execute_one(
    self,
    query: ClauseElement,
    values: Optional[Dict[str, Any]] = None,
    return_metadata: bool = False,
) -> Union[str, Dict[str, Any]]:
    """
            Executes a single SQL statement asynchronously and returns a meaningful result.

            Supports both read (SELECT) and write (INSERT/UPDATE/DELETE) statements.
            - For SELECT, returns the fetched records (list of scalars, dicts, or ORM objects).
            - For INSERT/UPDATE/DELETE, returns a dict with metadata including rowcount,
                inserted_primary_key (if available), and any returned rows when the statement
                includes RETURNING.

    Args:
        query (ClauseElement): An SQLAlchemy query object representing the SQL statement to execute.
        values (Optional[Dict[str, Any]]): A dictionary of parameter values to bind to the query.
            Defaults to None.
        return_metadata (bool): When True, returns metadata for DML statements.
            When False (default), returns "complete" for DML statements to preserve legacy behavior.

            Returns:
                    - For SELECT: List[Any] of records.
                    - For DML (INSERT/UPDATE/DELETE) when return_metadata=True: Dict with keys:
                            {
                                "rowcount": int | None,
                                "inserted_primary_key": List[Any] | None,
                                "rows": List[Any] (if statement returns rows)
                            }
                    - For DML when return_metadata=False: the string "complete".
                    - On error: Dict[str, str] from handle_exceptions.

    Example:
        ```python
        from sqlalchemy import insert

        query = insert(User).values(name='John Doe')
        result = await db_ops.execute_one(query)
        ```
    """
    logger.debug("Starting execute_one operation")
    try:
        async with self.async_db.get_db_session() as session:
            logger.debug(f"Executing query: {query}")
            result = await session.execute(query, params=values)

            # SELECT → shape and return records directly from this same
            # session/transaction; no commit needed for a pure read.
            if self._classify_statement(query) == "select":
                records = self._shape_rows_from_result(result)
                logger.debug(f"execute_one result: {records}")
                return records

            rows = self._collect_rows(result)
            meta = self._assemble_dml_metadata(query, result, rows)

            # Commit for non-SELECT statements (DML/DDL).
            await session.commit()

            logger.debug(f"Query executed successfully with metadata: {meta}")
            return meta if return_metadata else "complete"
    except Exception as ex:
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

get_columns_details(table) async

Retrieves the details of the columns of a given table.

This asynchronous method accepts a table object and returns a dictionary. Each key in the dictionary is a column name from the table, and the corresponding value is another dictionary containing details about that column, such as type, if it's nullable, if it's a primary key, if it's unique, its autoincrement status, and its default value.

Parameters:

Name	Type	Description	Default
table	Table	An instance of the SQLAlchemy Table class	required

Returns:

Name	Type	Description
dict		A dictionary where each key is a column name, and each value
		is a dictionary with the column's details.

Raises:

Type	Description
Exception	If any error occurs during the database operation.

Example

columns = await db_ops.get_columns_details(User)

Source code in dsg_lib/async_database_functions/database_operations.py

async def get_columns_details(self, table):
    """
    Retrieves the details of the columns of a given table.

    This asynchronous method accepts a table object and returns a
    dictionary. Each key in the dictionary is a column name from the table,
    and the corresponding value is another dictionary containing details
    about that column, such as type, if it's nullable, if it's a primary
    key, if it's unique, its autoincrement status, and its default value.

    Args:
        table (Table): An instance of the SQLAlchemy Table class
        representing the database table for which column details are
        required.

    Returns:
        dict: A dictionary where each key is a column name, and each value
        is a dictionary with the column's details.

    Raises:
        Exception: If any error occurs during the database operation.

    Example:
        ```python
        columns = await db_ops.get_columns_details(User)
        ```
    """
    # Log the start of the operation
    logger.debug(
        f"Starting get_columns_details operation for table: {table.__name__}"
    )

    try:
        # Log the start of the column retrieval
        logger.debug(f"Getting columns for table: {table.__name__}")

        # Retrieve the details of the columns and store them in a dictionary
        # The keys are the column names and the values are dictionaries
        # containing the column details
        columns = {
            c.name: {
                "type": str(c.type),
                "nullable": c.nullable,
                "primary_key": c.primary_key,
                "unique": c.unique,
                "autoincrement": c.autoincrement,
                "default": (
                    str(c.default.arg)
                    if c.default is not None and not callable(c.default.arg)
                    else None
                ),
            }
            for c in table.__table__.columns
        }

        # Log the successful column retrieval
        logger.debug(f"Successfully retrieved columns for table: {table.__name__}")

        return columns
    except Exception as ex:  # pragma: no cover
        # Handle any exceptions that occur during the column retrieval
        logger.error(
            f"An error occurred while getting columns for table: {table.__name__}"
        )  # pragma: no cover
        return handle_exceptions(ex)  # pragma: no cover

get_primary_keys(table) async

Retrieves the primary keys of a given table.

This asynchronous method accepts a table object and returns a list containing the names of its primary keys. It is useful for understanding the structure of the table and for operations that require knowledge of the primary keys.

Parameters:

Name	Type	Description	Default
table	Table	An instance of the SQLAlchemy Table class	required

Returns:

Name	Type	Description
list		A list containing the names of the primary keys of the table.

Raises:

Type	Description
Exception	If any error occurs during the database operation.

Example

primary_keys = await db_ops.get_primary_keys(User)

Source code in dsg_lib/async_database_functions/database_operations.py

async def get_primary_keys(self, table):
    """
    Retrieves the primary keys of a given table.

    This asynchronous method accepts a table object and returns a list
    containing the names of its primary keys. It is useful for understanding
    the structure of the table and for operations that require knowledge of
    the primary keys.

    Args:
        table (Table): An instance of the SQLAlchemy Table class
        representing the database table for which primary keys are required.

    Returns:
        list: A list containing the names of the primary keys of the table.

    Raises:
        Exception: If any error occurs during the database operation.

    Example:
        ```python
        primary_keys = await db_ops.get_primary_keys(User)
        ```
    """
    # Log the start of the operation
    logger.debug(f"Starting get_primary_keys operation for table: {table.__name__}")

    try:
        # Log the start of the primary key retrieval
        logger.debug(f"Getting primary keys for table: {table.__name__}")

        # Retrieve the primary keys and store them in a list
        primary_keys = table.__table__.primary_key.columns.keys()

        # Log the successful primary key retrieval
        logger.debug(f"Primary keys retrieved successfully: {primary_keys}")

        return primary_keys

    except Exception as ex:  # pragma: no cover
        # Handle any exceptions that occur during the primary key retrieval
        logger.error(f"Exception occurred: {ex}")  # pragma: no cover
        return handle_exceptions(ex)  # pragma: no cover

get_table_names() async

Retrieves the names of all tables in the database.

This asynchronous method returns a list containing the names of all tables in the database. It is useful for database introspection, allowing the user to know which tables are available in the current database context.

Returns:

Name	Type	Description
list		A list containing the names of all tables in the database.

Raises:

Type	Description
Exception	If any error occurs during the database operation.

Example

table_names = await db_ops.get_table_names()

Source code in dsg_lib/async_database_functions/database_operations.py

async def get_table_names(self):
    """
    Retrieves the names of all tables in the database.

    This asynchronous method returns a list containing the names of all
    tables in the database. It is useful for database introspection,
    allowing the user to know which tables are available in the current
    database context.

    Returns:
        list: A list containing the names of all tables in the database.

    Raises:
        Exception: If any error occurs during the database operation.

    Example:
        ```python
        table_names = await db_ops.get_table_names()
        ```
    """
    # Log the start of the operation
    logger.debug("Starting get_table_names operation")

    try:
        # Log the start of the table name retrieval
        logger.debug("Retrieving table names")

        # Retrieve the table names and store them in a list The keys of the
        # metadata.tables dictionary are the table names
        table_names = list(self.async_db.Base.metadata.tables.keys())

        # Log the successful table name retrieval
        logger.debug(f"Table names retrieved successfully: {table_names}")

        return table_names

    except Exception as ex:  # pragma: no cover
        # Handle any exceptions that occur during the table name retrieval
        logger.error(f"Exception occurred: {ex}")  # pragma: no cover
        return handle_exceptions(ex)  # pragma: no cover

paginate_query(query, page=1, page_size=100) async

Executes a query with page-based pagination, returning the requested page of records together with the total matching record count.

This combines what would otherwise be a separate count_query call and a read_query call (each opening their own session) into one method that counts and fetches the page within a single session.

Parameters:

Name	Type	Description	Default
query	Select	A SQLAlchemy Select query object (without its own .limit()/.offset() applied) specifying the records to page through. Apply any .where()/.order_by() clauses beforehand.	required
page	int	The 1-indexed page number to fetch. Defaults to 1.	1
page_size	int	The number of records per page. Defaults to 100.	100

Returns:

Name	Type	Description
dict	Union[Dict[str, Any], DatabaseErrorResult]	`{"items": List[Any], "total": int, "page": int,
	Union[Dict[str, Any], DatabaseErrorResult]	"page_size": int, "pages": int}, whereitems` is the shaped page
	Union[Dict[str, Any], DatabaseErrorResult]	of records (scalars, dicts, or ORM objects, as in read_query) and
	Union[Dict[str, Any], DatabaseErrorResult]	pages is the total number of pages (0 if there are no matches).
DatabaseErrorResult	Union[Dict[str, Any], DatabaseErrorResult]	If page/page_size are invalid, or an error
	Union[Dict[str, Any], DatabaseErrorResult]	occurs during the query.

Example

result = await db_ops.paginate_query(select(User).order_by(User.name), page=2, page_size=25)
# result["items"], result["total"], result["pages"]

Source code in dsg_lib/async_database_functions/database_operations.py

async def paginate_query(
    self,
    query,
    page: int = 1,
    page_size: int = 100,
) -> Union[Dict[str, Any], DatabaseErrorResult]:
    """
    Executes a query with page-based pagination, returning the requested
    page of records together with the total matching record count.

    This combines what would otherwise be a separate `count_query` call and
    a `read_query` call (each opening their own session) into one method
    that counts and fetches the page within a single session.

    Args:
        query (Select): A SQLAlchemy `Select` query object (without its own
            `.limit()`/`.offset()` applied) specifying the records to page
            through. Apply any `.where()`/`.order_by()` clauses beforehand.
        page (int): The 1-indexed page number to fetch. Defaults to 1.
        page_size (int): The number of records per page. Defaults to 100.

    Returns:
        dict: `{"items": List[Any], "total": int, "page": int,
        "page_size": int, "pages": int}`, where `items` is the shaped page
        of records (scalars, dicts, or ORM objects, as in `read_query`) and
        `pages` is the total number of pages (0 if there are no matches).
        DatabaseErrorResult: If `page`/`page_size` are invalid, or an error
        occurs during the query.

    Example:
        ```python
        result = await db_ops.paginate_query(select(User).order_by(User.name), page=2, page_size=25)
        # result["items"], result["total"], result["pages"]
        ```
    """
    if page < 1 or page_size < 1:
        return DatabaseErrorResult(
            {
                "error": "Invalid pagination parameters",
                "details": f"page and page_size must be >= 1 (got page={page}, page_size={page_size})",
            }
        )

    logger.debug(f"Starting paginate_query operation: page={page}, page_size={page_size}")

    try:
        async with self.async_db.get_db_session() as session:
            count_result = await session.execute(
                select(func.count()).select_from(query.subquery())
            )
            total = count_result.scalar()

            paged_query = query.limit(page_size).offset((page - 1) * page_size)
            result = await session.execute(paged_query)
            items = self._shape_rows_from_result(result)

            pages = (total + page_size - 1) // page_size if total else 0

            logger.debug(f"paginate_query result: total={total}, pages={pages}")
            return {
                "items": items,
                "total": total,
                "page": page,
                "page_size": page_size,
                "pages": pages,
            }

    except Exception as ex:
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

read_multi_query(queries) async

Executes multiple fetch queries asynchronously and returns a dictionary of results for each query.

This asynchronous method accepts a dictionary where each key is a query name (str) and each value is a SQLAlchemy Select query object. It executes each query within a single database session and collects the results. The results are returned as a dictionary mapping each query name to a list of records that match that query.

The function automatically determines the structure of each result set: - If the query returns a single column, the result will be a list of scalar values. - If the query returns multiple columns, the result will be a list of dictionaries mapping column names to values. - If the result row is an ORM object, it will be returned as-is.

Parameters:

Name	Type	Description	Default
queries	Dict[str, Select]	A dictionary mapping query names to SQLAlchemy Select query objects.	required

Returns:

Type	Description
	Dict[str, List[Any]]: A dictionary where each key is a query name and each value is a list of records
	(scalars, dictionaries, or ORM objects) that match the corresponding query.

Raises:

Type	Description
Exception	If any error occurs during the execution of any query, the function logs the error and

Example

from sqlalchemy import select
queries = {
    "adults": select(User).where(User.age >= 18),
    "minors": select(User).where(User.age < 18),
}
results = await db_ops.read_multi_query(queries)
# results["adults"] and results["minors"] will contain lists of records

Source code in dsg_lib/async_database_functions/database_operations.py

async def read_multi_query(self, queries: Dict[str, str]):
    """
    Executes multiple fetch queries asynchronously and returns a dictionary of results for each query.

    This asynchronous method accepts a dictionary where each key is a query name (str)
    and each value is a SQLAlchemy `Select` query object. It executes each query within a single
    database session and collects the results. The results are returned as a dictionary mapping
    each query name to a list of records that match that query.

    The function automatically determines the structure of each result set:
    - If the query returns a single column, the result will be a list of scalar values.
    - If the query returns multiple columns, the result will be a list of dictionaries mapping column names to values.
    - If the result row is an ORM object, it will be returned as-is.

    Args:
        queries (Dict[str, Select]): A dictionary mapping query names to SQLAlchemy `Select` query objects.

    Returns:
        Dict[str, List[Any]]: A dictionary where each key is a query name and each value is a list of records
        (scalars, dictionaries, or ORM objects) that match the corresponding query.

    Raises:
        Exception: If any error occurs during the execution of any query, the function logs the error and
        returns a dictionary with error details using `handle_exceptions`.

    Example:
        ```python
        from sqlalchemy import select
        queries = {
            "adults": select(User).where(User.age >= 18),
            "minors": select(User).where(User.age < 18),
        }
        results = await db_ops.read_multi_query(queries)
        # results["adults"] and results["minors"] will contain lists of records
        ```
    """
    # Log the start of the operation
    logger.debug("Starting read_multi_query operation")

    try:
        results = {}
        async with self.async_db.get_db_session() as session:
            for query_name, query in queries.items():
                logger.debug(f"Executing fetch query: {query}")
                result = await session.execute(query)
                results[query_name] = self._shape_rows_from_result(result)
        return results

    except Exception as ex:
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

read_one_record(query) async

Retrieves a single record from the database based on the provided query.

This asynchronous method accepts a SQL query object and returns the first record that matches the query. If no record matches the query, it returns None. This method is useful for fetching specific data when the expected result is a single record.

Parameters:

Name	Type	Description	Default
query	Select	An instance of the SQLAlchemy Select class,	required

Returns:

Name	Type	Description
Result		The first record that matches the query or None if no record matches.

Raises:

Type	Description
Exception	If any error occurs during the database operation.

Example

record = await db_ops.read_one_record(select(User).where(User.name == 'John Doe'))

Source code in dsg_lib/async_database_functions/database_operations.py

async def read_one_record(self, query):
    """
    Retrieves a single record from the database based on the provided query.

    This asynchronous method accepts a SQL query object and returns the
    first record that matches the query. If no record matches the query, it
    returns None. This method is useful for fetching specific data
    when the expected result is a single record.

    Parameters:
        query (Select): An instance of the SQLAlchemy Select class,
        representing the query to be executed.

    Returns:
        Result: The first record that matches the query or None if no record matches.

    Raises:
        Exception: If any error occurs during the database operation.

    Example:
        ```python
        record = await db_ops.read_one_record(select(User).where(User.name == 'John Doe'))
        ```
    """
    # Log the start of the operation
    logger.debug(f"Starting read_one_record operation for {query}")

    try:
        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the start of the record retrieval
            logger.debug(f"Getting record with query: {query}")

            # Execute the query and retrieve the first record
            result = await session.execute(query)
            record = result.scalar_one()

            # Log the successful record retrieval
            logger.debug(f"Record retrieved successfully: {record}")

            return record

    except NoResultFound:
        # No record was found
        logger.debug("No record found")
        return None

    except Exception as ex:  # pragma: no cover
        # Handle any exceptions that occur during the record retrieval
        logger.error(f"Exception occurred: {ex}")  # pragma: no cover
        return handle_exceptions(ex)  # pragma: no cover

update_one(table, record_id, new_values) async

This method is deprecated. Use execute_one with an UPDATE query instead.

Updates a single record in the database identified by its ID.

This asynchronous method takes a SQLAlchemy Table object, a record ID, and a dictionary of new values to update the record. It updates the specified record in the given table with the new values. The method does not allow updating certain fields, such as 'id' or 'date_created'.

Parameters:

Name	Type	Description	Default
table	Table	The SQLAlchemy Table object representing the table	required
in the database. record_id	str	The ID of the record to be	required
updated. new_values	dict	A dictionary containing the fields to	required

Returns:

Name	Type	Description
Base		The updated record if successful; otherwise, an error
		dictionary.

Raises:

Type	Description
Exception	If any error occurs during the update operation.

Example

record = await db_ops.update_one(User, 1, {'name': 'John Smith'})

Source code in dsg_lib/async_database_functions/database_operations.py

@deprecated("Use `execute_one` with a UPDATE query instead.")
async def update_one(self, table, record_id: str, new_values: dict):
    """
    This method is deprecated. Use `execute_one` with an UPDATE query instead.

    Updates a single record in the database identified by its ID.

    This asynchronous method takes a SQLAlchemy `Table` object, a record ID,
    and a dictionary of new values to update the record. It updates the
    specified record in the given table with the new values. The method does
    not allow updating certain fields, such as 'id' or 'date_created'.

    Parameters:
        table (Table): The SQLAlchemy `Table` object representing the table
        in the database. record_id (str): The ID of the record to be
        updated. new_values (dict): A dictionary containing the fields to
        update and their new values.

    Returns:
        Base: The updated record if successful; otherwise, an error
        dictionary.

    Raises:
        Exception: If any error occurs during the update operation.

    Example:
        ```python
        record = await db_ops.update_one(User, 1, {'name': 'John Smith'})
        ```
    """
    non_updatable_fields = ["id", "date_created"]

    # Log the start of the operation
    logger.debug(
        f"Starting update_one operation for record_id: {record_id} in table: {table.__name__}"
    )

    try:
        # Start a new database session
        async with self.async_db.get_db_session() as session:
            # Log the record being fetched
            logger.debug(f"Fetching record with id: {record_id}")

            # Fetch the record
            record = await session.get(table, record_id)
            if not record:
                # Log the error if no record is found
                logger.error(f"No record found with pkid: {record_id}")
                return DatabaseErrorResult(
                    {
                        "error": "Record not found",
                        "details": f"No record found with pkid {record_id}",
                    }
                )

            # Log the record being updated
            logger.debug(f"Updating record with new values: {new_values}")

            # Update the record with the new values
            for key, value in new_values.items():
                if key not in non_updatable_fields:
                    setattr(record, key, value)
            await session.commit()

            # Log the successful record update
            logger.debug(f"Record updated successfully: {record.pkid}")
            return record

    except Exception as ex:
        # Handle any exceptions that occur during the record update
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

handle_exceptions(ex)

Handles exceptions for database operations.

This function checks the type of the exception, logs an appropriate error message, and returns a DatabaseErrorResult containing the error details.

Parameters:

Name	Type	Description	Default
ex	Exception	The exception to handle.	required

Returns:

Name	Type	Description
DatabaseErrorResult	DatabaseErrorResult	A dict-like object with two keys: 'error' and 'details'.

Example:

try:
    await db_ops.read_query(select(User))
except Exception as ex:
    error_details = handle_exceptions(ex)

Source code in dsg_lib/async_database_functions/database_operations.py

def handle_exceptions(ex: Exception) -> DatabaseErrorResult:
    """
    Handles exceptions for database operations.

    This function checks the type of the exception, logs an appropriate error
    message, and returns a `DatabaseErrorResult` containing the error details.

    Args:
        ex (Exception): The exception to handle.

    Returns:
        DatabaseErrorResult: A dict-like object with two keys: 'error' and 'details'.

    Example:
    ```python
    try:
        await db_ops.read_query(select(User))
    except Exception as ex:
        error_details = handle_exceptions(ex)
    ```
    """
    # Extract the error message before the SQL statement
    error_only = str(ex).split("[SQL:")[0]

    # Check the type of the exception
    if isinstance(ex, IntegrityError):
        # Log the error and return the error details
        logger.error(f"IntegrityError occurred: {ex}")
        return DatabaseErrorResult({"error": "IntegrityError", "details": error_only})
    elif isinstance(ex, SQLAlchemyError):
        # Log the error and return the error details
        logger.error(f"SQLAlchemyError occurred: {ex}")
        return DatabaseErrorResult({"error": "SQLAlchemyError", "details": error_only})
    else:
        # Log the error and return the error details
        logger.error(f"Exception occurred: {ex}")
        return DatabaseErrorResult({"error": "General Exception", "details": str(ex)})