DevSetGo Library

Description	Common functions for Python applications. This is to increase reusability and limit rewritting the same functions in multiple applications. It also allows for defects to be addressed quickly and then be propigated across applications.
Author(s)	Mike Ryan
Repository	https://github.com/devsetgo/dsg_lib
Copyright	Copyright © 2016 - 2024 Mike Ryan

Support Python Versions

CI/CD Pipeline:

SonarCloud:

DevSetGo Common Library

devsetgo_lib is a versatile library designed to provide common functions for Python applications. Its main goal is to increase reusability and reduce the need to rewrite the same functions across multiple applications. This also allows for quick defect resolution and propagation of fixes across all dependent projects.

Read the Full Documentation here.

Key Features

Common Functions:

• File Operations:

  • CSV, JSON, and Text File Functions: Create, read, write, and manipulate various file types with ease.
  • Folder Functions: Create and remove directories, list directory contents, and manage file system operations efficiently.

• Logging: Comprehensive logging setup using the Loguru Library. Provides extensive customization options for log configuration, including log rotation, retention, and formatting. Includes improvements for multiprocessing environments to ensure log messages are handled correctly across multiple processes.

• Calendar Functions: Convert between month names and numbers seamlessly.

• Pattern Matching: Powerful tools for searching patterns in text using regular expressions.

FastAPI Endpoints:

• Pre-built endpoints for system health checks, status, and uptime monitoring.
• Functions to generate HTTP response codes easily.

Async Database:

• Configuration and management of asynchronous database sessions.
• CRUD operations with async support.

---

Installation

To install devsetgo_lib, use pip:

pip install devsetgo-lib

# For async database setup with SQLite or PostgreSQL
pip install devsetgo-lib[sqlite]
pip install devsetgo-lib[postgres]

# Experimental support for other databases
pip install devsetgo-lib[oracle]
pip install devsetgo-lib[mssql]
pip install devsetgo-lib[mysql]

# For adding FastAPI endpoints
pip install devsetgo-lib[fastapi]

# Install everything
pip install devsetgo-lib[all]

Usage

Here's a quick example to demonstrate how you can use some of the key features of devsetgo_lib:

from devsetgo_lib.common_functions import file_functions, logging_config, patterns, calendar_functions

# File Operations
file_functions.create_sample_files("example", 100)
content = file_functions.read_from_file("example.csv")
print(content)

# Logging
logging_config.config_log(logging_directory='logs', log_name='app.log', logging_level='DEBUG')
logger = logging.getLogger('app_logger')
logger.info("This is an info message")

# Pattern Matching
text = "Hello, my name is 'John Doe' and I live in 'New York'."
results = patterns.pattern_between_two_char(text, "'", "'")
print(results)

# Calendar Functions
print(calendar_functions.get_month(1))  # Output: 'January'
print(calendar_functions.get_month_number('January'))  # Output: 1

For detailed documentation on each module and function, please refer to the official documentation.

Contributing

We welcome contributions! Please see our contributing guidelines for more details.

License

This project is licensed under the MIT License. See the LICENSE file for more details.

Contact

For any questions or issues, please open an issue on GitHub or contact us at devsetgo@example.com.

Quick Start

Install

pip install devsetgo-lib

# Aysync database setup
pip install devsetgo-lib[sqlite]
pip install devsetgo-lib[postgres]

# Consider these experimental and untested
pip install devsetgo-lib[oracle]
pip install devsetgo-lib[mssql]
pip install devsetgo-lib[mysql]

# For adding FastAPI endpoints
pip install devsetgo-lib[fastapi]

# Install everything
pip install devsetgo-lib[all]

See documentation for more examples of library use

Common Functions

Reference

dsg_lib.common_functions.logging_config

This module provides a comprehensive logging setup using the loguru library, facilitating easy logging management for Python applications.

The config_log function, central to this module, allows for extensive customization of logging behavior. It supports specifying the logging directory, log file name, logging level, and controls for log rotation, retention, and formatting among other features. Additionally, it offers advanced options like backtrace and diagnose for in-depth debugging, and the ability to append the application name to the log file for clearer identification.

Usage example:

from dsg_lib.common_functions.logging_config import config_log

config_log(
    logging_directory='logs',  # Directory for storing logs
    log_name='log',  # Base name for log files
    logging_level='DEBUG',  # Minimum logging level
    log_rotation='100 MB',  # Size threshold for log rotation
    log_retention='30 days',  # Duration to retain old log files
    enqueue=True,  # Enqueue log messages
)

# Example log messages
logger.debug("This is a debug message")
logger.info("This is an info message")
logger.error("This is an error message")
logger.warning("This is a warning message")
logger.critical("This is a critical message")

Todo

• Add support for additional logging handlers.
• Implement asynchronous logging.

Author

Mike Ryan

Date Created

2021/07/16

Date Updated

2024/07/27

License

MIT

SafeFileSink

A class to handle safe file logging with rotation and retention policies.

This class provides mechanisms to manage log files by rotating them based on size and retaining them for a specified duration. It also supports optional compression of log files.

Attributes:

Name	Type	Description
path	str	The path to the log file.
rotation_size	int	The size threshold for log rotation in bytes.
retention_days	timedelta	The duration to retain old log files.
compression	str	The compression method to use for old log files.

Methods:

Name	Description
parse_size	Parses a size string (e.g., '100MB') and returns the size in bytes.
parse_duration	Parses a duration string (e.g., '7 days') and returns a timedelta object.

Example

safe_file_sink = SafeFileSink( path='logs/app.log', rotation='100 MB', retention='30 days', compression='zip' )

This will set up a log file at 'logs/app.log' with rotation at 100 MB,

retention for 30 days, and compression using zip.

Source code in dsg_lib/common_functions/logging_config.py

class SafeFileSink:
    """
    A class to handle safe file logging with rotation and retention policies.

    This class provides mechanisms to manage log files by rotating them based on size and retaining them for a specified duration. It also supports optional compression of log files.

    Attributes:
        path (str): The path to the log file.
        rotation_size (int): The size threshold for log rotation in bytes.
        retention_days (timedelta): The duration to retain old log files.
        compression (str, optional): The compression method to use for old log files.

    Methods:
        parse_size(size_str): Parses a size string (e.g., '100MB') and returns the size in bytes.
        parse_duration(duration_str): Parses a duration string (e.g., '7 days') and returns a timedelta object.

    Example:
        safe_file_sink = SafeFileSink(
            path='logs/app.log',
            rotation='100 MB',
            retention='30 days',
            compression='zip'
        )

        # This will set up a log file at 'logs/app.log' with rotation at 100 MB,
        # retention for 30 days, and compression using zip.
    """

    def __init__(self, path, rotation, retention, compression=None):
        self.path = path
        self.rotation_size = self.parse_size(rotation)
        self.retention_days = self.parse_duration(retention)
        self.compression = compression

    @staticmethod
    def parse_size(size_str):  # pragma: no cover
        """
        Parses a size string and returns the size in bytes.

        Args:
            size_str (str): The size string (e.g., '100MB').

        Returns:
            int: The size in bytes.
        """
        size_str = size_str.upper()
        if size_str.endswith("MB"):
            return int(size_str[:-2]) * 1024 * 1024
        elif size_str.endswith("GB"):
            return int(size_str[:-2]) * 1024 * 1024 * 1024
        elif size_str.endswith("KB"):
            return int(size_str[:-2]) * 1024
        else:
            return int(size_str)

    @staticmethod
    def parse_duration(duration_str):  # pragma: no cover
        """
        Parses a duration string and returns a timedelta object.

        Args:
            duration_str (str): The duration string (e.g., '7 days').

        Returns:
            timedelta: The duration as a timedelta object.
        """
        duration_str = duration_str.lower()
        if "day" in duration_str:
            return timedelta(days=int(duration_str.split()[0]))
        elif "hour" in duration_str:
            return timedelta(hours=int(duration_str.split()[0]))
        elif "minute" in duration_str:
            return timedelta(minutes=int(duration_str.split()[0]))
        else:
            return timedelta(days=0)

    def __call__(self, message):  # pragma: no cover
        """
        Handles the logging of a message, including writing, rotating, and applying retention policies.

        Args:
            message (str): The log message to be written.

        This method ensures thread-safe logging by acquiring a lock before writing the message,
        rotating the logs if necessary, and applying the retention policy to remove old log files.
        """
        with rotation_lock:
            self.write_message(message)
            self.rotate_logs()
            self.apply_retention()

    def write_message(self, message):  # pragma: no cover
        """
        Writes a log message to the log file.

        Args:
            message (str): The log message to be written.

        This method opens the log file in append mode and writes the message to it.
        """
        with open(self.path, "a") as f:
            f.write(message)

    def rotate_logs(self):  # pragma: no cover
        """
        Rotates the log file if it exceeds the specified rotation size.

        This method checks the size of the current log file. If the file size exceeds the specified rotation size, it renames the current log file by appending a timestamp to its name. Optionally, it compresses the rotated log file using the specified compression method and removes the original uncompressed file.

        Args:
            None

        Returns:
            None

        Raises:
            OSError: If there is an error renaming or compressing the log file.
        """
        if os.path.getsize(self.path) >= self.rotation_size:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            rotated_path = f"{self.path}.{timestamp}"
            os.rename(self.path, rotated_path)
            if self.compression:
                shutil.make_archive(
                    rotated_path,
                    self.compression,
                    root_dir=os.path.dirname(rotated_path),
                    base_dir=os.path.basename(rotated_path),
                )
                os.remove(rotated_path)

    def apply_retention(self):  # pragma: no cover
        """
        Applies the retention policy to remove old log files.

        This method iterates through the log files in the directory of the current log file. It checks the modification time of each log file and removes those that are older than the specified retention period.

        Args:
            None

        Returns:
            None

        Raises:
            OSError: If there is an error removing a log file.
        """
        now = datetime.now()
        for filename in os.listdir(os.path.dirname(self.path)):
            if (
                filename.startswith(os.path.basename(self.path))
                and len(filename.split(".")) > 1
            ):
                file_path = os.path.join(os.path.dirname(self.path), filename)
                file_time = datetime.fromtimestamp(os.path.getmtime(file_path))
                if now - file_time > self.retention_days:
                    os.remove(file_path)

__call__(message)

Handles the logging of a message, including writing, rotating, and applying retention policies.

Parameters:

Name	Type	Description	Default
message	str	The log message to be written.	required

This method ensures thread-safe logging by acquiring a lock before writing the message, rotating the logs if necessary, and applying the retention policy to remove old log files.

Source code in dsg_lib/common_functions/logging_config.py

def __call__(self, message):  # pragma: no cover
    """
    Handles the logging of a message, including writing, rotating, and applying retention policies.

    Args:
        message (str): The log message to be written.

    This method ensures thread-safe logging by acquiring a lock before writing the message,
    rotating the logs if necessary, and applying the retention policy to remove old log files.
    """
    with rotation_lock:
        self.write_message(message)
        self.rotate_logs()
        self.apply_retention()

apply_retention()

Applies the retention policy to remove old log files.

This method iterates through the log files in the directory of the current log file. It checks the modification time of each log file and removes those that are older than the specified retention period.

Returns:

Type	Description
	None

Raises:

Type	Description
OSError	If there is an error removing a log file.

Source code in dsg_lib/common_functions/logging_config.py

def apply_retention(self):  # pragma: no cover
    """
    Applies the retention policy to remove old log files.

    This method iterates through the log files in the directory of the current log file. It checks the modification time of each log file and removes those that are older than the specified retention period.

    Args:
        None

    Returns:
        None

    Raises:
        OSError: If there is an error removing a log file.
    """
    now = datetime.now()
    for filename in os.listdir(os.path.dirname(self.path)):
        if (
            filename.startswith(os.path.basename(self.path))
            and len(filename.split(".")) > 1
        ):
            file_path = os.path.join(os.path.dirname(self.path), filename)
            file_time = datetime.fromtimestamp(os.path.getmtime(file_path))
            if now - file_time > self.retention_days:
                os.remove(file_path)

parse_duration(duration_str) staticmethod

Parses a duration string and returns a timedelta object.

Parameters:

Name	Type	Description	Default
duration_str	str	The duration string (e.g., '7 days').	required

Returns:

Name	Type	Description
timedelta		The duration as a timedelta object.

Source code in dsg_lib/common_functions/logging_config.py

@staticmethod
def parse_duration(duration_str):  # pragma: no cover
    """
    Parses a duration string and returns a timedelta object.

    Args:
        duration_str (str): The duration string (e.g., '7 days').

    Returns:
        timedelta: The duration as a timedelta object.
    """
    duration_str = duration_str.lower()
    if "day" in duration_str:
        return timedelta(days=int(duration_str.split()[0]))
    elif "hour" in duration_str:
        return timedelta(hours=int(duration_str.split()[0]))
    elif "minute" in duration_str:
        return timedelta(minutes=int(duration_str.split()[0]))
    else:
        return timedelta(days=0)

parse_size(size_str) staticmethod

Parses a size string and returns the size in bytes.

Parameters:

Name	Type	Description	Default
size_str	str	The size string (e.g., '100MB').	required

Returns:

Name	Type	Description
int		The size in bytes.

Source code in dsg_lib/common_functions/logging_config.py

@staticmethod
def parse_size(size_str):  # pragma: no cover
    """
    Parses a size string and returns the size in bytes.

    Args:
        size_str (str): The size string (e.g., '100MB').

    Returns:
        int: The size in bytes.
    """
    size_str = size_str.upper()
    if size_str.endswith("MB"):
        return int(size_str[:-2]) * 1024 * 1024
    elif size_str.endswith("GB"):
        return int(size_str[:-2]) * 1024 * 1024 * 1024
    elif size_str.endswith("KB"):
        return int(size_str[:-2]) * 1024
    else:
        return int(size_str)

rotate_logs()

Rotates the log file if it exceeds the specified rotation size.

This method checks the size of the current log file. If the file size exceeds the specified rotation size, it renames the current log file by appending a timestamp to its name. Optionally, it compresses the rotated log file using the specified compression method and removes the original uncompressed file.

Returns:

Type	Description
	None

Raises:

Type	Description
OSError	If there is an error renaming or compressing the log file.

Source code in dsg_lib/common_functions/logging_config.py

def rotate_logs(self):  # pragma: no cover
    """
    Rotates the log file if it exceeds the specified rotation size.

    This method checks the size of the current log file. If the file size exceeds the specified rotation size, it renames the current log file by appending a timestamp to its name. Optionally, it compresses the rotated log file using the specified compression method and removes the original uncompressed file.

    Args:
        None

    Returns:
        None

    Raises:
        OSError: If there is an error renaming or compressing the log file.
    """
    if os.path.getsize(self.path) >= self.rotation_size:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        rotated_path = f"{self.path}.{timestamp}"
        os.rename(self.path, rotated_path)
        if self.compression:
            shutil.make_archive(
                rotated_path,
                self.compression,
                root_dir=os.path.dirname(rotated_path),
                base_dir=os.path.basename(rotated_path),
            )
            os.remove(rotated_path)

write_message(message)

Writes a log message to the log file.

Parameters:

Name	Type	Description	Default
message	str	The log message to be written.	required

This method opens the log file in append mode and writes the message to it.

Source code in dsg_lib/common_functions/logging_config.py

def write_message(self, message):  # pragma: no cover
    """
    Writes a log message to the log file.

    Args:
        message (str): The log message to be written.

    This method opens the log file in append mode and writes the message to it.
    """
    with open(self.path, "a") as f:
        f.write(message)

config_log(logging_directory='log', log_name='log', logging_level='INFO', log_rotation='100 MB', log_retention='30 days', log_backtrace=False, log_format=None, log_serializer=False, log_diagnose=False, app_name=None, append_app_name=False, enqueue=True, intercept_standard_logging=True, compression='zip')

Configures the logging settings for the application.

This function sets up the logging configuration, including the log directory, log file name, logging level, log rotation, retention policies, and other optional settings.

Parameters:

Name	Type	Description	Default
logging_directory	str	The directory where log files will be stored. Defaults to "log".	'log'
log_name	str	The base name of the log file. Defaults to "log".	'log'
logging_level	str	The logging level (e.g., "INFO", "DEBUG"). Defaults to "INFO".	'INFO'
log_rotation	str	The size threshold for log rotation (e.g., "100 MB"). Defaults to "100 MB".	'100 MB'
log_retention	str	The duration to retain old log files (e.g., "30 days"). Defaults to "30 days".	'30 days'
log_backtrace	bool	Whether to include backtrace information in logs. Defaults to False.	False
log_format	str	The format string for log messages. Defaults to a predefined format if not provided.	None
log_serializer	bool	Whether to serialize log messages. Defaults to False.	False
log_diagnose	bool	Whether to include diagnostic information in logs. Defaults to False.	False
app_name	str	The name of the application. Defaults to None.	None
append_app_name	bool	Whether to append the application name to the log file name. Defaults to False.	False
enqueue	bool	Whether to enqueue log messages for asynchronous processing. Defaults to True.	True
intercept_standard_logging	bool	Whether to intercept standard logging calls. Defaults to True.	True
compression	str	The compression method for rotated log files (e.g., "zip"). Defaults to 'zip'.	'zip'

Returns:

Type	Description
	None

Example

config_log( logging_directory='logs', log_name='app_log', logging_level='DEBUG', log_rotation='50 MB', log_retention='7 days', log_backtrace=True, log_format='{time} - {level} - {message}', log_serializer=True, log_diagnose=True, app_name='MyApp', append_app_name=True, enqueue=False, intercept_standard_logging=False, compression='gz' )

This will configure the logging settings with the specified parameters, setting up a log file at 'logs/app_log' with rotation at 50 MB, retention for 7 days, and other specified options.

Source code in dsg_lib/common_functions/logging_config.py

def config_log(
    logging_directory: str = "log",
    log_name: str = "log",
    logging_level: str = "INFO",
    log_rotation: str = "100 MB",
    log_retention: str = "30 days",
    log_backtrace: bool = False,
    log_format: str = None,
    log_serializer: bool = False,
    log_diagnose: bool = False,
    app_name: str = None,
    append_app_name: bool = False,
    enqueue: bool = True,
    intercept_standard_logging: bool = True,
    compression: str = "zip",
):
    """
    Configures the logging settings for the application.

    This function sets up the logging configuration, including the log directory, log file name, logging level, log rotation, retention policies, and other optional settings.

    Args:
        logging_directory (str): The directory where log files will be stored. Defaults to "log".
        log_name (str): The base name of the log file. Defaults to "log".
        logging_level (str): The logging level (e.g., "INFO", "DEBUG"). Defaults to "INFO".
        log_rotation (str): The size threshold for log rotation (e.g., "100 MB"). Defaults to "100 MB".
        log_retention (str): The duration to retain old log files (e.g., "30 days"). Defaults to "30 days".
        log_backtrace (bool): Whether to include backtrace information in logs. Defaults to False.
        log_format (str, optional): The format string for log messages. Defaults to a predefined format if not provided.
        log_serializer (bool): Whether to serialize log messages. Defaults to False.
        log_diagnose (bool): Whether to include diagnostic information in logs. Defaults to False.
        app_name (str, optional): The name of the application. Defaults to None.
        append_app_name (bool): Whether to append the application name to the log file name. Defaults to False.
        enqueue (bool): Whether to enqueue log messages for asynchronous processing. Defaults to True.
        intercept_standard_logging (bool): Whether to intercept standard logging calls. Defaults to True.
        compression (str): The compression method for rotated log files (e.g., "zip"). Defaults to 'zip'.

    Returns:
        None

    Example:
        config_log(
            logging_directory='logs',
            log_name='app_log',
            logging_level='DEBUG',
            log_rotation='50 MB',
            log_retention='7 days',
            log_backtrace=True,
            log_format='{time} - {level} - {message}',
            log_serializer=True,
            log_diagnose=True,
            app_name='MyApp',
            append_app_name=True,
            enqueue=False,
            intercept_standard_logging=False,
            compression='gz'
        )

    This will configure the logging settings with the specified parameters, setting up a log file at 'logs/app_log' with rotation at 50 MB, retention for 7 days, and other specified options.
    """

    # If the log_name ends with ".log", remove the extension
    if log_name.endswith(".log"):
        log_name = log_name.replace(".log", "")  # pragma: no cover

    # If the log_name ends with ".json", remove the extension
    if log_name.endswith(".json"):
        log_name = log_name.replace(".json", "")  # pragma: no cover

    # Set default log format if not provided
    if log_format is None:  # pragma: no cover
        log_format = "<green>{time:YYYY-MM-DD HH:mm:ss.SSSSSS}</green> | <level>{level: <8}</level> | <cyan> {name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"  # pragma: no cover

    if log_serializer is True:
        log_format = "{message}"  # pragma: no cover
        log_name = f"{log_name}.json"  # pragma: no cover
    else:
        log_name = f"{log_name}.log"  # pragma: no cover

    # Validate logging level
    log_levels: list = ["DEBUG", "INFO", "ERROR", "WARNING", "CRITICAL"]
    if logging_level.upper() not in log_levels:
        raise ValueError(
            f"Invalid logging level: {logging_level}. Valid levels are: {log_levels}"
        )

    # Generate unique trace ID
    trace_id: str = str(uuid4())
    logger.configure(extra={"app_name": app_name, "trace_id": trace_id})

    # Append app name to log format if provided
    if app_name is not None:
        log_format += " | app_name: {extra[app_name]}"

    # Remove any previously added sinks
    logger.remove()

    # Append app name to log file name if required
    if append_app_name is True and app_name is not None:
        log_name = log_name.replace(".", f"_{app_name}.")

    # Construct log file path
    log_path = Path.cwd().joinpath(logging_directory).joinpath(log_name)

    # Add loguru logger with specified configuration
    logger.add(
        SafeFileSink(
            log_path,
            rotation=log_rotation,
            retention=log_retention,
            compression=compression,
        ),
        level=logging_level.upper(),
        format=log_format,
        enqueue=enqueue,
        backtrace=log_backtrace,
        serialize=log_serializer,
        diagnose=log_diagnose,
    )

    basic_config_handlers: list = []

    class InterceptHandler(logging.Handler):
        """
        A logging handler that intercepts standard logging messages and redirects them to Loguru.

        This handler captures log messages from the standard logging module and forwards them to Loguru, preserving the log level and message details.

        Methods:
            emit(record):
                Emits a log record to Loguru.
        """

        def emit(self, record):
            """
            Emits a log record to Loguru.

            This method captures the log record, determines the appropriate Loguru log level, and logs the message using Loguru. It also handles exceptions and finds the caller's frame to maintain accurate log information.

            Args:
                record (logging.LogRecord): The log record to be emitted.

            Returns:
                None
            """
            # Get corresponding Loguru level if it exists
            try:
                level = logger.level(record.levelname).name
            except ValueError:  # pragma: no cover
                level = record.levelno  # pragma: no cover

            # Find caller from where originated the logged message
            frame, depth = logging.currentframe(), 2
            while frame.f_code.co_filename == logging.__file__:  # pragma: no cover
                frame = frame.f_back  # pragma: no cover
                depth += 1  # pragma: no cover

            # Log the message using loguru
            logger.opt(depth=depth, exception=record.exc_info).log(
                level, record.getMessage()
            )  # pragma: no cover

    if intercept_standard_logging:
        # Add interceptor handler to all existing loggers
        for name in logging.Logger.manager.loggerDict:
            logging.getLogger(name).addHandler(InterceptHandler())

        # Add interceptor handler to the root logger
        basic_config_handlers.append(InterceptHandler())

    # Set the root logger's level to the lowest level possible
    logging.getLogger().setLevel(logging.NOTSET)

    if intercept_standard_logging:
        # Append an InterceptHandler instance to the handlers list if intercept_standard_logging is True
        basic_config_handlers.append(InterceptHandler())

    if len(basic_config_handlers) > 0:
        logging.basicConfig(handlers=basic_config_handlers, level=logging_level.upper())

Reference

dsg_lib.common_functions.file_functions

file_functions.py

This module provides a function to delete a file with a specified name from a specified directory.

Functions:

Name

Description

delete_file

str) -> str: Deletes a file with the specified file name from the directory specified by the directory_to_files variable. The file type is determined by the file extension, and the file is deleted from the subdirectory corresponding to the file type. Args: file_name (str): The name of the file to be deleted. Returns: str: A string indicating that the file has been deleted. Raises: TypeError: If the file name is not a string. ValueError: If the file name contains a forward slash or backslash, or if the file type is not supported. FileNotFoundError: If the file does not exist.

Example:

from dsg_lib.common_functions import file_functions

file_functions.delete_file("test.csv")

# Outputs: 'complete'

Author: Mike Ryan Date: 2024/05/16 License: MIT

create_sample_files(file_name, sample_size)

Create sample CSV and JSON files with random data.

Parameters:

Name	Type	Description	Default
file_name	str	The base name for the sample files (without extension).	required
sample_size	int	The number of rows to generate for the sample files.	required

Returns:

Type	Description
None	None

Raises:

Type	Description
Exception	If an error occurs while creating the sample files.

Example: ```python from dsg_lib.common_functions import file_functions

file_functions.create_sample_files("test", 100)

Creates 'test.csv' and 'test.json' each with 100 rows of random data ```

Source code in dsg_lib/common_functions/file_functions.py

def create_sample_files(file_name: str, sample_size: int) -> None:
    """
    Create sample CSV and JSON files with random data.

    Args:
        file_name (str): The base name for the sample files (without extension).
        sample_size (int): The number of rows to generate for the sample files.

    Returns:
        None

    Raises:
        Exception: If an error occurs while creating the sample files.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions

    file_functions.create_sample_files("test", 100)

    # Creates 'test.csv' and 'test.json' each with 100 rows of random data ```
    """
    logger.debug(f"Creating sample files for {file_name} with {sample_size} rows.")

    try:
        # Generate the CSV data
        csv_header = ["name", "birth_date", "number"]
        csv_data: List[List[str]] = [csv_header]

        # Generate rows for CSV data
        for i in range(1, sample_size + 1):
            r_int: int = random.randint(0, len(first_name) - 1)
            name = first_name[r_int]
            row: List[str] = [name, generate_random_date(), str(i)]
            csv_data.append(row)

        # Save the CSV file
        csv_file = f"{file_name}.csv"
        save_csv(csv_file, csv_data)

        # Generate the JSON data
        json_data: List[dict] = []

        # Generate rows for JSON data
        for _ in range(1, sample_size + 1):
            r_int: int = random.randint(0, len(first_name) - 1)
            name = first_name[r_int]
            sample_dict: dict = {
                "name": name,
                "birthday_date": generate_random_date(),
            }
            json_data.append(sample_dict)

        # Save the JSON file
        json_file: str = f"{file_name}.json"
        save_json(json_file, json_data)

        # Log the data
        logger.debug(f"CSV Data: {csv_data}")
        logger.debug(f"JSON Data: {json_data}")

    except Exception as e:  # pragma: no cover
        logger.exception(
            f"Error occurred while creating sample files: {e}"
        )  # pragma: no cover
        raise  # pragma: no cover

delete_file(file_name)

Deletes a file with the specified file name from the specified directory. The file type is determined by the file extension.

Parameters:

Name	Type	Description	Default
directory_to_files	str	The directory where the file is located.	required
file_name	str	The name of the file to be deleted.	required

Returns:

Name	Type	Description
str	str	A message indicating whether the file has been deleted successfully
	str	or an error occurred.

Raises:

Type	Description
TypeError	If the directory or file name is not a string. ValueError: If
is not supported. FileNotFoundError	If the file does not exist.

Example:

from dsg_lib.common_functions import file_functions

file_functions.delete_file("test.csv")

# Outputs: 'File deleted successfully'

Source code in dsg_lib/common_functions/file_functions.py

def delete_file(file_name: str) -> str:
    """
    Deletes a file with the specified file name from the specified directory.
    The file type is determined by the file extension.

    Args:
        directory_to_files (str): The directory where the file is located.
        file_name (str): The name of the file to be deleted.

    Returns:
        str: A message indicating whether the file has been deleted successfully
        or an error occurred.

    Raises:
        TypeError: If the directory or file name is not a string. ValueError: If
        the file name contains a forward slash or backslash, or if the file type
        is not supported. FileNotFoundError: If the file does not exist.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions

    file_functions.delete_file("test.csv")

    # Outputs: 'File deleted successfully'
    ```
    """
    logger.info(f"Deleting file: {file_name}")

    # Check that the file name is a string
    if not isinstance(file_name, str):
        raise TypeError(f"{file_name} is not a valid string")

    # Split the file name into its name and extension components
    file_name, file_ext = os.path.splitext(file_name)

    # Check that the file name does not contain a forward slash or backslash
    if os.path.sep in file_name:
        raise ValueError(f"{file_name} cannot contain {os.path.sep}")

    # Check that the file type is supported
    if file_ext not in directory_map:
        raise ValueError(
            f"unsupported file type: {file_ext}. Supported file types are: {', '.join(directory_map.keys())}"
        )

    # Construct the full file path
    file_directory = Path.cwd() / directory_to_files / directory_map[file_ext]
    file_path = file_directory / f"{file_name}{file_ext}"

    # Check that the file exists
    if not file_path.is_file():
        raise FileNotFoundError(f"file not found: {file_name}{file_ext}")

    # Delete the file
    os.remove(file_path)
    logger.info(f"File {file_name}{file_ext} deleted from file path: {file_path}")

    # Return a string indicating that the file has been deleted
    return "complete"

generate_random_date()

Generate a random datetime string in the format yyyy-mm-dd hh:mm:ss.ffffff.

Returns:

Name	Type	Description
str	str	A randomly generated datetime string.

Example:

from dsg_lib.common_functions import file_functions
random_date = file_functions.generate_random_date()
# Returns: '1992-03-15 10:30:45.123456'

Source code in dsg_lib/common_functions/file_functions.py

def generate_random_date() -> str:
    """
    Generate a random datetime string in the format yyyy-mm-dd hh:mm:ss.ffffff.

    Returns:
        str: A randomly generated datetime string.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions
    random_date = file_functions.generate_random_date()
    # Returns: '1992-03-15 10:30:45.123456'
    ```
    """
    # Define the minimum and maximum years for the date range
    min_year: int = 1905
    max_year: int = datetime.now().year

    # Generate random values for the year, month, day, hour, minute, and second
    year: int = random.randrange(min_year, max_year + 1)
    month: int = random.randint(1, 12)
    day: int = random.randint(1, 28)
    hour: int = random.randint(0, 12)
    minute: int = random.randint(0, 59)
    second: int = random.randint(0, 59)

    # Create a datetime object with the random values
    date_value: datetime = datetime(year, month, day, hour, minute, second)

    # Format the datetime string and return it
    return f"{date_value:%Y-%m-%d %H:%M:%S.%f}"

open_csv(file_name, delimiter=',', quote_level='minimal', skip_initial_space=True)

Opens a CSV file with the specified file name and returns its contents as a list of dictionaries.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the file to open. Should include the '.csv'	required
extension.	delimiter (str	The character used to separate	required
fields	in the CSV file. Defaults to ','. quote_level (str		required
optional)		Whether to skip initial whitespace in the CSV file. Defaults	required

Returns:

Name	Type	Description
list	list	The contents of the CSV file as a list of dictionaries. Each
	list	dictionary represents a row in the CSV file, where the keys are column
	list	names and the values are the data for those columns.

Raises:

Type	Description
TypeError	If file_name is not a string. ValueError: If quote_level
is not a valid level. FileNotFoundError	If the file does not exist.

Example:

from dsg_lib.common_functions import file_functions data =
file_functions.open_csv("test.csv", delimiter=";", quote_level="all",
skip_initial_space=False)  # Returns: [{'column1': 'value1', 'column2':
'value2'}]

Source code in dsg_lib/common_functions/file_functions.py

def open_csv(
    file_name: str,
    delimiter: str = ",",
    quote_level: str = "minimal",
    skip_initial_space: bool = True,
) -> list:
    """
    Opens a CSV file with the specified file name and returns its contents as a
    list of dictionaries.

    Args:
        file_name (str): The name of the file to open. Should include the '.csv'
        extension. delimiter (str, optional): The character used to separate
        fields in the CSV file. Defaults to ','. quote_level (str, optional):
        The quoting level used in the CSV file. Valid levels are "none",
        "minimal", and "all". Defaults to "minimal". skip_initial_space (bool,
        optional): Whether to skip initial whitespace in the CSV file. Defaults
        to True.

    Returns:
        list: The contents of the CSV file as a list of dictionaries. Each
        dictionary represents a row in the CSV file, where the keys are column
        names and the values are the data for those columns.

    Raises:
        TypeError: If `file_name` is not a string. ValueError: If `quote_level`
        is not a valid level. FileNotFoundError: If the file does not exist.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions data =
    file_functions.open_csv("test.csv", delimiter=";", quote_level="all",
    skip_initial_space=False)  # Returns: [{'column1': 'value1', 'column2':
    'value2'}]
    ```
    """
    # A dictionary that maps quote levels to csv quoting constants
    quote_levels = {
        "none": csv.QUOTE_NONE,
        "minimal": csv.QUOTE_MINIMAL,
        "all": csv.QUOTE_ALL,
    }

    # Check that file name is a string
    if not isinstance(file_name, str):
        error = f"{file_name} is not a valid string"
        logger.error(error)
        raise TypeError(error)

    # Check that quote level is valid
    quote_level = quote_level.lower()
    if quote_level not in quote_levels:
        error = f"Invalid quote level: {quote_level}. Valid levels are: {', '.join(quote_levels)}"
        logger.error(error)
        raise ValueError(error)
    quoting = quote_levels[quote_level]

    # Add extension to file name and create file path
    file_name = f"{file_name}.csv"
    file_directory = Path.cwd().joinpath(directory_to_files).joinpath("csv")
    file_path = file_directory.joinpath(file_name)

    # Check that file exists
    if not file_path.is_file():
        error = f"File not found: {file_path}"
        logger.error(error)
        raise FileNotFoundError(error)

    # Read CSV file
    data = []
    with file_path.open(encoding="utf-8") as f:
        reader = csv.DictReader(
            f,
            delimiter=delimiter,
            quoting=quoting,
            skipinitialspace=skip_initial_space,
        )
        for row in reader:
            data.append(dict(row))

    logger.info(f"File opened: {file_name}")
    return data

open_json(file_name)

Open a JSON file and load its contents into a dictionary.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the JSON file to open.	required

Returns:

Name	Type	Description
dict	dict	The contents of the JSON file as a dictionary.

Raises:

Type	Description
TypeError	If the file name is not a string. FileNotFoundError: If the

Source code in dsg_lib/common_functions/file_functions.py

def open_json(file_name: str) -> dict:
    """
    Open a JSON file and load its contents into a dictionary.

    Args:
        file_name (str): The name of the JSON file to open.

    Returns:
        dict: The contents of the JSON file as a dictionary.

    Raises:
        TypeError: If the file name is not a string. FileNotFoundError: If the
        file does not exist.
    """
    # Check if file name is a string
    if not isinstance(file_name, str):
        error = f"{file_name} is not a valid string"
        logger.error(error)
        raise TypeError(error)

    file_directory = Path(directory_to_files) / directory_map[".json"]
    file_save = file_directory / file_name

    # Check if path correct
    if not file_save.is_file():
        error = f"file not found error: {file_save}"
        logger.exception(error)
        raise FileNotFoundError(error)

    # open file
    with open(file_save) as read_file:
        # load file into data variable
        result = json.load(read_file)

    logger.info(f"File Opened: {file_name}")
    return result

open_text(file_name)

Opens a text file with the specified file name and returns its contents as a string.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the file to open. Should include the '.txt'	required

Returns:

Name	Type	Description
str	str	The contents of the text file as a string.

Raises:

Type	Description
TypeError	If the file_name parameter is not a string or contains a
forward slash. FileNotFoundError	If the file does not exist.

Example:

from dsg_lib.common_functions import file_functions
data = file_functions.open_text("test.txt")
# Returns: 'This is a test text file.'

Source code in dsg_lib/common_functions/file_functions.py

def open_text(file_name: str) -> str:
    """
    Opens a text file with the specified file name and returns its contents as a
    string.

    Args:
        file_name (str): The name of the file to open. Should include the '.txt'
        extension.

    Returns:
        str: The contents of the text file as a string.

    Raises:
        TypeError: If the `file_name` parameter is not a string or contains a
        forward slash. FileNotFoundError: If the file does not exist.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions
    data = file_functions.open_text("test.txt")
    # Returns: 'This is a test text file.'
    ```
    """
    # Replace backslashes with forward slashes in the file name
    if "\\" in file_name:  # pragma: no cover
        file_name = file_name.replace("\\", "/")  # pragma: no cover

    # Check that file_name does not contain invalid characters
    if "/" in file_name:
        logger.error(f"{file_name} cannot contain /")
        raise TypeError(f"{file_name} cannot contain /")

    # Get the path to the text directory and the file path
    file_directory = os.path.join(directory_to_files, "text")
    file_path = Path.cwd().joinpath(file_directory, file_name)

    # Check if the file exists
    if not file_path.is_file():
        raise FileNotFoundError(f"file not found error: {file_path}")

    # Open the file and read the data
    with open(file_path, "r", encoding="utf-8") as file:
        data = file.read()

    logger.info(f"File opened: {file_path}")
    return data

save_csv(file_name, data, root_folder=None, delimiter=',', quotechar='"')

Saves a list of dictionaries as a CSV file with the specified file name in the specified directory. Each dictionary in the list should represent a row in the CSV file.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the file to save the data in. Should	required
include	the '.csv' extension. data (list	The data to be saved. Each	required
optional)		The root directory where the file will be saved. If None, the	required
(str,	optional	The character used to separate fields in the CSV file.	required
Defaults	to ','. quotechar (str	The character used to quote	required

Returns:

Name	Type	Description
str	str	A message indicating whether the file has been saved successfully
	str	or an error occurred.

Raises:

Type	Description
TypeError	If the data is not a list, or the file name, delimiter, or
quotechar is not a string. ValueError	If the file name does not end

Example: ```python from dsg_lib.common_functions import file_functions

data = [{"column1": "value1", "column2": "value2"}]

file_functions.save_csv("test.csv", data, "/path/to/directory", delimiter=";", quotechar="'")

Saves data to '/path/to/directory/test.csv'

```

Source code in dsg_lib/common_functions/file_functions.py

def save_csv(
    file_name: str,
    data: list,
    root_folder: str = None,
    delimiter: str = ",",
    quotechar: str = '"',
) -> str:
    """
    Saves a list of dictionaries as a CSV file with the specified file name in
    the specified directory. Each dictionary in the list should represent a row
    in the CSV file.

    Args:
        file_name (str): The name of the file to save the data in. Should
        include the '.csv' extension. data (list): The data to be saved. Each
        element of the list should be a dictionary where the keys are column
        names and the values are the data for those columns. root_folder (str,
        optional): The root directory where the file will be saved. If None, the
        file will be saved in the current directory. Defaults to None. delimiter
        (str, optional): The character used to separate fields in the CSV file.
        Defaults to ','. quotechar (str, optional): The character used to quote
        fields in the CSV file. Defaults to '"'.

    Returns:
        str: A message indicating whether the file has been saved successfully
        or an error occurred.

    Raises:
        TypeError: If the data is not a list, or the file name, delimiter, or
        quotechar is not a string. ValueError: If the file name does not end
        with '.csv'.

    Example: ```python
    from dsg_lib.common_functions import file_functions

    data = [{"column1": "value1", "column2": "value2"}]

    file_functions.save_csv("test.csv", data, "/path/to/directory", delimiter=";", quotechar="'")

    # Saves data to '/path/to/directory/test.csv'
    ```
    """
    # Set the root folder to directory_to_files if None
    if root_folder is None:
        root_folder = directory_to_files

    # Create the csv directory if it does not exist
    csv_directory = Path(root_folder) / "csv"
    csv_directory.mkdir(parents=True, exist_ok=True)

    # Check that delimiter and quotechar are single characters
    if len(delimiter) != 1:
        raise TypeError(f"{delimiter} can only be a single character")
    if len(quotechar) != 1:
        raise TypeError(f"{quotechar} can only be a single character")

    # Check that data is a list
    if not isinstance(data, list):
        raise TypeError(f"{data} is not a valid list")

    # Check that file_name is a string and does not contain invalid characters
    if not isinstance(file_name, str) or "/" in file_name or "\\" in file_name:
        raise TypeError(f"{file_name} is not a valid file name")

    # Add extension to file_name if needed
    if not file_name.endswith(".csv"):
        file_name += ".csv"

    # Create the file path
    file_path = csv_directory / file_name

    # Write data to file
    with open(file_path, "w", encoding="utf-8", newline="") as csv_file:
        csv_writer = csv.writer(csv_file, delimiter=delimiter, quotechar=quotechar)
        csv_writer.writerows(data)

    logger.info(f"File Create: {file_name}")
    return "complete"

save_json(file_name, data, root_folder=None)

Saves a dictionary or a list as a JSON file with the specified file name in the specified directory.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the file to save the data in. Should	required
include	the '.json' extension. data (list or dict	The data to be	required
saved.	root_folder (str	The root directory where the file	required

Returns:

Name	Type	Description
str	str	A message indicating whether the file has been saved successfully
	str	or an error occurred.

Raises:

Type	Description
TypeError	If the data is not a list or a dictionary, or the file name
or directory is not a string. ValueError	If the file name contains a

Example:

from dsg_lib.common_functions import file_functions

data = {"key": "value"}

file_functions.save_json("test.json", data, "/path/to/directory")

# Saves data to '/path/to/directory/test.json'

Source code in dsg_lib/common_functions/file_functions.py

def save_json(file_name: str, data, root_folder: str = None) -> str:
    """
    Saves a dictionary or a list as a JSON file with the specified file name in
    the specified directory.

    Args:
        file_name (str): The name of the file to save the data in. Should
        include the '.json' extension. data (list or dict): The data to be
        saved. root_folder (str, optional): The root directory where the file
        will be saved. Defaults to None, which means the file will be saved in
        the 'data' directory.

    Returns:
        str: A message indicating whether the file has been saved successfully
        or an error occurred.

    Raises:
        TypeError: If the data is not a list or a dictionary, or the file name
        or directory is not a string. ValueError: If the file name contains a
        forward slash or backslash, or if the file name does not end with
        '.json'.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions

    data = {"key": "value"}

    file_functions.save_json("test.json", data, "/path/to/directory")

    # Saves data to '/path/to/directory/test.json'
    ```
    """
    try:
        # Validate inputs
        if not isinstance(data, (list, dict)):
            raise TypeError(
                f"data must be a list or a dictionary instead of type {type(data)}"
            )
        if "/" in file_name or "\\" in file_name:
            raise ValueError(f"{file_name} cannot contain / or \\")

        # Add extension if not present in file_name
        if not file_name.endswith(".json"):  # pragma: no cover
            file_name += ".json"  # pragma: no cover

        if root_folder is None:
            root_folder = directory_to_files

        # Determine directory
        json_directory = Path(root_folder) / "json"

        # Construct file path
        file_path = json_directory / file_name

        # Create the json directory if it does not exist
        json_directory.mkdir(parents=True, exist_ok=True)

        # Write data to file
        with open(file_path, "w") as write_file:
            json.dump(data, write_file)

        # Log success message
        logger.info(f"File created: {file_path}")

        return "File saved successfully"

    except (TypeError, ValueError) as e:
        logger.error(f"Error creating file {file_name}: {e}")
        raise

save_text(file_name, data, root_folder=None)

Saves a string of text to a file with the specified file name in the specified directory.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the file to save the data in. Should not	required
include	the '.txt' extension. data (str	The text data to be saved.	required
root_folder	str	The root directory where the file will be	None

Returns:

Name	Type	Description
str	str	A message indicating whether the file has been saved successfully
	str	or an error occurred.

Raises:

Type	Description
TypeError	If the data parameter is not a string, or the file_name
contains a forward slash or backslash. FileNotFoundError	If the

Example:

from dsg_lib.common_functions import file_functions

file_functions.save_text("test", "This is a test text file.", "/path/to/directory")

# Saves data to '/path/to/directory/test.txt'

Source code in dsg_lib/common_functions/file_functions.py

def save_text(file_name: str, data: str, root_folder: str = None) -> str:
    """
    Saves a string of text to a file with the specified file name in the
    specified directory.

    Args:
        file_name (str): The name of the file to save the data in. Should not
        include the '.txt' extension. data (str): The text data to be saved.
        root_folder (str, optional): The root directory where the file will be
        saved. If None, the file will be saved in the current directory.
        Defaults to None.

    Returns:
        str: A message indicating whether the file has been saved successfully
        or an error occurred.

    Raises:
        TypeError: If the `data` parameter is not a string, or the `file_name`
        contains a forward slash or backslash. FileNotFoundError: If the
        directory does not exist.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions

    file_functions.save_text("test", "This is a test text file.", "/path/to/directory")

    # Saves data to '/path/to/directory/test.txt'
    ```
    """
    # If no root folder is provided, use the default directory
    if root_folder is None:  # pragma: no cover
        root_folder = directory_to_files  # pragma: no cover

    # Determine the directory for text files
    text_directory = Path(root_folder) / "text"

    # Construct the file path for text files
    file_path = text_directory / file_name

    # Create the text directory if it does not exist
    text_directory.mkdir(parents=True, exist_ok=True)

    # Check that data is a string and that file_name does not contain invalid
    # characters
    if not isinstance(data, str):
        logger.error(f"{file_name} is not a valid string")
        raise TypeError(f"{file_name} is not a valid string")
    elif "/" in file_name or "\\" in file_name:
        logger.error(f"{file_name} cannot contain \\ or /")
        raise ValueError(f"{file_name} cannot contain \\ or /")

    # Add extension to file_name if needed
    if not file_name.endswith(".txt"):
        file_name += ".txt"
    # Open or create the file and write the data
    with open(file_path, "w+", encoding="utf-8") as file:
        file.write(data)

    logger.info(f"File created: {file_path}")
    return "complete"

Reference

dsg_lib.common_functions.folder_functions

This module contains functions for working with directories and files.

Functions:

Name	Description
last_data_files_changed	Get the last modified file in a
get_directory_list	Get a list of directories in the
specified directory. make_folder	Make a folder in a
specific directory. remove_folder	Remove a folder from the

Example:

from dsg_lib.common_functions import folder_functions

# Get the last modified file in a directory time_stamp, file_path =
folder_functions.last_data_files_changed("/path/to/directory")  # Returns:
(datetime.datetime(2022, 1, 1, 12, 0, 0), '/path/to/directory/test.txt')

# Get a list of directories in the specified directory directories =
folder_functions.get_directory_list("/path/to/directory")  # Returns:
['/path/to/directory/dir1', '/path/to/directory/dir2']

# Make a folder in a specific directory
folder_functions.make_folder("/path/to/directory/new_folder")  # Creates a new
folder at '/path/to/directory/new_folder'

# Remove a folder from the specified directory
folder_functions.remove_folder("/path/to/directory/old_folder")  # Removes the
folder at '/path/to/directory/old_folder'

Author: Mike Ryan Date: 2024/05/16 License: MIT

get_directory_list(file_directory)

Get a list of directories in the specified directory.

Parameters:

Name	Type	Description	Default
file_directory	str	The path of the directory to check.	required

Returns:

Type	Description
List[str]	List[str]: A list of directories in the specified directory.

Raises:

Type	Description
FileNotFoundError	If the directory does not exist.

Example:

from dsg_lib import file_functions

directories = file_functions.get_directory_list("/path/to/directory")

# Returns: ['/path/to/directory/dir1', '/path/to/directory/dir2']

Source code in dsg_lib/common_functions/folder_functions.py

def get_directory_list(file_directory: str) -> List[str]:
    """
    Get a list of directories in the specified directory.

    Args:
        file_directory (str): The path of the directory to check.

    Returns:
        List[str]: A list of directories in the specified directory.

    Raises:
        FileNotFoundError: If the directory does not exist.

    Example:
    ```python
    from dsg_lib import file_functions

    directories = file_functions.get_directory_list("/path/to/directory")

    # Returns: ['/path/to/directory/dir1', '/path/to/directory/dir2']
    ```
    """
    # Create a Path object for the specified directory
    file_path = Path.cwd().joinpath(file_directory)

    try:
        # Use a list comprehension to create a list of directories in the
        # specified directory
        direct_list = [x for x in file_path.iterdir() if x.is_dir()]

        # Log a message indicating that the list of directories was retrieved
        logger.info(f"Retrieved list of directories: {file_directory}")

        # Return the list of directories
        return direct_list

    except FileNotFoundError as err:
        # Log an error message if the specified directory does not exist
        logger.error(err)

last_data_files_changed(directory_path)

Get the last modified file in a directory and return its modification time and path.

Parameters:

Name	Type	Description	Default
directory_path	str	The path of the directory to check.	required

Returns:

Type	Description
datetime	Tuple[datetime, str]: A tuple containing the modification time and path
str	of the last modified file.

Raises:

Type	Description
FileNotFoundError	If the directory does not exist.

Example:

from dsg_lib import file_functions

time_stamp, file_path = file_functions.last_data_files_changed("/path/to/directory")

# Returns: (datetime.datetime(2022, 1, 1, 12, 0, 0), '/path/to/directory/test.txt')

Source code in dsg_lib/common_functions/folder_functions.py

def last_data_files_changed(directory_path: str) -> Tuple[datetime, str]:
    """
    Get the last modified file in a directory and return its modification time
    and path.

    Args:
        directory_path (str): The path of the directory to check.

    Returns:
        Tuple[datetime, str]: A tuple containing the modification time and path
        of the last modified file.

    Raises:
        FileNotFoundError: If the directory does not exist.

    Example:
    ```python
    from dsg_lib import file_functions

    time_stamp, file_path = file_functions.last_data_files_changed("/path/to/directory")

    # Returns: (datetime.datetime(2022, 1, 1, 12, 0, 0), '/path/to/directory/test.txt')
    ```
    """
    try:
        # Use a generator expression to find the last modified file in the
        # directory
        time, file_path = max((f.stat().st_mtime, f) for f in directory_path.iterdir())

        # Convert the modification time to a datetime object
        time_stamp = datetime.fromtimestamp(time)

        # Log a message to indicate that the directory was checked for the last
        # modified file
        logger.info(f"Directory checked for last change: {directory_path}")

        # Return the modification time and path of the last modified file
        return time_stamp, file_path

    except Exception as err:
        # Log an error message if an exception occurs, and return a default
        # value to indicate an error
        logger.error(err)
        return None, None

remove_folder(file_directory)

Remove a folder from the specified directory.

Parameters:

Name	Type	Description	Default
file_directory	str	The directory containing the folder to be removed.	required

Returns:

Type	Description
None	None.

Raises:

Type	Description
FileNotFoundError	If the specified directory does not exist. OSError:

Example:

from dsg_lib.common_functions import file_functions

file_functions.remove_folder("/path/to/directory/old_folder")

# Removes the folder at '/path/to/directory/old_folder'

Source code in dsg_lib/common_functions/folder_functions.py

def remove_folder(file_directory: str) -> None:
    """
    Remove a folder from the specified directory.

    Args:
        file_directory (str): The directory containing the folder to be removed.

    Returns:
        None.

    Raises:
        FileNotFoundError: If the specified directory does not exist. OSError:
        If the specified folder could not be removed.

    Example:
    ```python
    from dsg_lib.common_functions import file_functions

    file_functions.remove_folder("/path/to/directory/old_folder")

    # Removes the folder at '/path/to/directory/old_folder'
    ```
    """
    try:
        # Create a Path object for the specified directory
        path = Path(file_directory)

        # Use the rmdir method of the Path object to remove the folder
        path.rmdir()

        # Log a message indicating that the folder was removed
        logger.info(f"Folder removed: {file_directory}")

    except FileNotFoundError as err:
        # Log an error message if the specified directory does not exist
        logger.error(err)

        # Raise the FileNotFoundError exception to be handled by the calling
        # code
        raise

    except OSError as err:
        # Log an error message if the folder could not be removed
        logger.error(err)

        # Raise the OSError exception to be handled by the calling code
        raise

Reference

dsg_lib.common_functions.patterns

This module contains functions for pattern searching in text using regular expressions.

The main function in this module is pattern_between_two_char, which searches for all patterns between two characters in a given string. The function uses Python's built-in re module for regex searching and the loguru module for logging.

Functions:

Name	Description
pattern_between_two_char	str, left_characters: str,
right_characters	str) -> dict: Searches for all patterns between two characters (left and right) in a given string using regular expressions.

Example

from dsg_lib.common_functions import patterns

text = "Hello, my name is 'John Doe' and I live in 'New York'." left_char =
"'" right_char = "'"

results = patterns.pattern_between_two_char(text, left_char, right_char)

print(results) ``` This will output: ```python {
    'found': ['John Doe', 'New York'], 'matched_found': 2,
    'pattern_parameters': {
        'left_character': "'", 'right_character': "'", 'regex_pattern':
        "'(.+?)'", 'text_string': "Hello, my name is 'John Doe' and I live
        in 'New York'."
    }
}

Author: Mike Ryan Date: 2024/05/16 License: MIT

pattern_between_two_char(text_string, left_characters, right_characters)

Searches for all patterns between two characters (left and right) in a given string using regular expressions.

This function takes a string and two characters as input, and returns a dictionary containing all patterns found between the two characters in the string. The dictionary also includes the number of matches found and the regex pattern used for searching.

The function uses Python's built-in re module for regex searching and the loguru module for logging.

Parameters:

Name	Type	Description	Default
text_string	str	The string in which to search for patterns.	required
left_characters	str	The character(s) that appear(s) immediately to	required
the	left of the desired pattern. right_characters (str	The	required

Returns:

Name	Type	Description
dict	dict	A dictionary with the following keys: - "found": a list of strings containing all patterns found. - "matched_found": the number of patterns found. - "pattern_parameters": a dictionary with the following keys: - "left_character": the escaped left character string used to build the regex pattern. - "right_character": the escaped right character string used to build the regex pattern. - "regex_pattern": the final regex pattern used for searching. - "text_string": the escaped input string used for searching.

Example

from dsg_lib.common_functions import patterns

text = "Hello, my name is 'John Doe' and I live in 'New York'."
left_char = "'" right_char = "'"

results = patterns.pattern_between_two_char(text, left_char, right_char)

print(results) ``` This will output: ```python {
    'found': ['John Doe', 'New York'], 'matched_found': 2,
    'pattern_parameters': {
        'left_character': "'", 'right_character': "'", 'regex_pattern':
        "'(.+?)'", 'text_string': "Hello, my name is 'John Doe' and I
        live in 'New York'."
    }
}

Source code in dsg_lib/common_functions/patterns.py

def pattern_between_two_char(
    text_string: str, left_characters: str, right_characters: str
) -> dict:
    """
    Searches for all patterns between two characters (left and right) in a given
    string using regular expressions.

    This function takes a string and two characters as input, and returns a
    dictionary containing all patterns found between the two characters in the
    string. The dictionary also includes the number of matches found and the
    regex pattern used for searching.

    The function uses Python's built-in `re` module for regex searching and the
    `loguru` module for logging.

    Args:
        text_string (str): The string in which to search for patterns.
        left_characters (str): The character(s) that appear(s) immediately to
        the left of the desired pattern. right_characters (str): The
        character(s) that appear(s) immediately to the right of the desired
        pattern.

    Returns:
        dict: A dictionary with the following keys:
            - "found": a list of strings containing all patterns found.
            - "matched_found": the number of patterns found.
            - "pattern_parameters": a dictionary with the following keys:
                - "left_character": the escaped left character string used to
                  build the regex pattern.
                - "right_character": the escaped right character string used to
                  build the regex pattern.
                - "regex_pattern": the final regex pattern used for searching.
                - "text_string": the escaped input string used for searching.

    Example:
        ```python
        from dsg_lib.common_functions import patterns

        text = "Hello, my name is 'John Doe' and I live in 'New York'."
        left_char = "'" right_char = "'"

        results = patterns.pattern_between_two_char(text, left_char, right_char)

        print(results) ``` This will output: ```python {
            'found': ['John Doe', 'New York'], 'matched_found': 2,
            'pattern_parameters': {
                'left_character': "'", 'right_character': "'", 'regex_pattern':
                "'(.+?)'", 'text_string': "Hello, my name is 'John Doe' and I
                live in 'New York'."
            }
        }
        ```
    """

    if not left_characters or not right_characters:
        raise ValueError(
            f"Left '{left_characters}' and/or Right '{right_characters}' characters must not be None or empty"
        )

    try:
        # Escape input strings to safely use them in regex pattern
        esc_text = re.escape(text_string)
        esc_left_char = re.escape(left_characters)
        esc_right_char = re.escape(right_characters)

        # Create a regex pattern that matches all substrings between target
        # characters
        pattern = f"{esc_left_char}(.+?){esc_right_char}"

        # Replace \w with . to match any printable UTF-8 character
        pattern = pattern.replace(r"\w", r".")

        # Search for all patterns and store them in pattern_list variable
        pattern_list = re.findall(pattern, esc_text)

        # Create a dictionary to store match details
        results: dict = {
            "found": pattern_list,
            "matched_found": len(pattern_list),
            "pattern_parameters": {
                "left_character": esc_left_char,
                "right_character": esc_right_char,
                "regex_pattern": pattern,
                "text_string": esc_text,
            },
        }

        # Log matched pattern(s) found using 'debug' log level
        if len(pattern_list) > 0:
            logger.debug(f"Matched pattern(s): {pattern_list}")

        # Log successful function execution using 'info' log level
        logger.info("Successfully executed 'pattern_between_two_char' function")
        return results

    except ValueError as e:  # pragma: no cover
        # capture exception and return error in case of invalid input parameters
        results: dict = {
            "error": str(e),
            "matched_found": 0,
            "pattern_parameters": {
                "left_character": left_characters,
                "right_character": right_characters,
                "regex_pattern": None,
                "text_string": text_string,
            },
        }
        # logger of regex error using 'critical' log level
        logger.critical(f"Failed to generate regex pattern with error: {e}")
        return results

Reference

dsg_lib.common_functions.calendar_functions

This module provides two main functions to convert between month numbers and their corresponding names.

Functions:

Name	Description
get_month	int) -> str: Converts an integer month number to its corresponding month name. Args: month (int): An integer between 1 and 12 representing the month number. Returns: str: The full name of the month corresponding to the input month number. If the input is not within the range of 1-12, returns "Invalid month number". If the input is not an integer, returns "Invalid input, integer is required".
get_month_number	str) -> int: Converts a month name to its corresponding month number. Args: month_name (str): A string containing the full name of a month. Returns: int: The month number corresponding to the input month name. If the input is not a valid month name, returns -1. If the input is not a string, returns "Invalid input, string is required".

Example:

from dsg_lib.common_functions.calendar_functions import get_month,

get_month_number print(get_month(1))

# Outputs: 'January'

print(get_month_number('January'))

# Outputs: 1

This module is part of the dsg_lib package and is used for handling and converting between month numbers and names.

Author: Mike Ryan Date: 2024/05/16 License: MIT

get_month(month)

Converts an integer month number to its corresponding month name.

Parameters:

Name	Type	Description	Default
month	int	An integer or integer-like float between 1 and 12	required

Returns:

Name	Type	Description
str	str	The full name of the month corresponding to the input month number. If the input is not within the range of 1-12, returns "Invalid month number". If the input is not an integer or integer-like float, returns "Invalid input, integer is required".

Source code in dsg_lib/common_functions/calendar_functions.py

def get_month(month: int) -> str:
    """
    Converts an integer month number to its corresponding month name.

    Args:
        month (int): An integer or integer-like float between 1 and 12
        representing the month number.

    Returns:
        str: The full name of the month corresponding to the input month number.
             If the input is not within the range of 1-12, returns "Invalid
             month number". If the input is not an integer or integer-like
             float, returns "Invalid input, integer is required".
    """

    # Define a tuple containing the names of all months
    months = (
        "January",
        "February",
        "March",
        "April",
        "May",
        "June",
        "July",
        "August",
        "September",
        "October",
        "November",
        "December",
    )

    # Convert integer-like floats to integers
    if isinstance(month, float) and month.is_integer():
        month = int(month)

    # Check if the input month is an integer
    if not isinstance(month, int):
        logger.error("Invalid input: %s, integer is required", month)
        return "Invalid input, integer is required"

    # Check if the input month is within the range of 1-12
    if 1 <= month <= 12:
        logger.info("Returning month name for month number: %s", month)
        return months[month - 1]
    else:
        logger.error(
            "Invalid input: %s, month number should be between 1 and 12", month
        )
        return "Invalid month number"

get_month_number(month_name)

Converts a month name to its corresponding month number.

Parameters:

Name	Type	Description	Default
month_name	str	A string containing the full name of a month.	required

Returns:

Name	Type	Description
int	int	The month number corresponding to the input month name. If the input is not a valid month name or not a string, returns -1.

Source code in dsg_lib/common_functions/calendar_functions.py

def get_month_number(month_name: str) -> int:
    """
    Converts a month name to its corresponding month number.

    Args:
        month_name (str): A string containing the full name of a month.

    Returns:
        int: The month number corresponding to the input month name.
             If the input is not a valid month name or not a string, returns -1.
    """

    # Define a dictionary mapping month names to month numbers
    month_dict = {
        "January": 1,
        "February": 2,
        "March": 3,
        "April": 4,
        "May": 5,
        "June": 6,
        "July": 7,
        "August": 8,
        "September": 9,
        "October": 10,
        "November": 11,
        "December": 12,
    }

    # Check if the input month name is a string
    if not isinstance(month_name, str):
        logger.error("Invalid input, string is required")
        return -1

    # Convert the input string to title case and remove leading/trailing spaces
    month_name = month_name.strip().title()

    # Check if the input month name is a valid key in the dictionary
    if month_name in month_dict:
        return month_dict[month_name]
    else:
        logger.error("Invalid month name: %s", month_name)
        return -1

Ended: Common Functions

FastAPI Functions

Reference

dsg_lib.fastapi_functions.http_codes

http_codes.py

This module provides a dictionary of HTTP status codes and their descriptions.

The dictionary ALL_HTTP_CODES contains the HTTP status codes as keys. Each key maps to another dictionary that contains a description of the status code, an extended description, and a link to its documentation on the Mozilla Developer Network (MDN).

Example:

from dsg_lib.fastapi_functions import http_codes

# Get the description, extended description, and link for HTTP status code 200
status_200 = http_codes.ALL_HTTP_CODES[200]
print(status_200)
# {'description': 'OK', 'extended_description': 'The request has succeeded', 'link': 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/200'}

Attributes:

Name	Type	Description
ALL_HTTP_CODES	dict	A dictionary of HTTP status codes. Each key is an

Author: Mike Ryan Date: 2024/05/16 License: MIT

DELETE_CODES = generate_code_dict(common_codes + [202, 204, 205, 409]) module-attribute

DELETE_CODES is a dictionary of HTTP status codes for DELETE requests. It includes all the common codes, plus some additional codes that are specific to DELETE requests.

Example:

from dsg_lib.fastapi_functions import http_codes

# Print the dictionary of HTTP status codes for DELETE requests
print(http_codes.DELETE_CODES)

GET_CODES = generate_code_dict(common_codes + [206, 304, 307, 410, 502]) module-attribute

GET_CODES is a dictionary of HTTP status codes for GET requests. It includes all the common codes, plus some additional codes that are specific to GET requests.

Example:

from dsg_lib.fastapi_functions import http_codes

# Print the dictionary of HTTP status codes for GET requests
print(http_codes.GET_CODES)

PATCH_CODES = generate_code_dict(common_codes + [202, 204, 206, 409, 412, 413]) module-attribute

PATCH_CODES is a dictionary of HTTP status codes for PATCH requests. It includes all the common codes, plus some additional codes that are specific to PATCH requests.

Example:

from dsg_lib.fastapi_functions import http_codes

# Print the dictionary of HTTP status codes for PATCH requests
print(http_codes.PATCH_CODES)

POST_CODES = generate_code_dict(common_codes + [201, 202, 205, 307, 409, 413, 415]) module-attribute

POST_CODES is a dictionary of HTTP status codes for POST requests. It includes all the common codes, plus some additional codes that are specific to POST requests.

Example:

from dsg_lib.fastapi_functions import http_codes

# Print the dictionary of HTTP status codes for POST requests
print(http_codes.POST_CODES)

PUT_CODES = generate_code_dict(common_codes + [202, 204, 206, 409, 412, 413]) module-attribute

PUT_CODES is a dictionary of HTTP status codes for PUT requests. It includes all the common codes, plus some additional codes that are specific to PUT requests.

Example:

from dsg_lib.fastapi_functions import http_codes

# Print the dictionary of HTTP status codes for PUT requests
print(http_codes.PUT_CODES)

generate_code_dict(codes, description_only=False)

Generate a dictionary of specific HTTP error codes from the http_codes dictionary.

This function takes a list of HTTP status codes and an optional boolean flag. If the flag is True, the function returns a dictionary where each key is an HTTP status code from the input list and each value is the corresponding description from the ALL_HTTP_CODES dictionary. If the flag is False, the function returns a dictionary where each key is an HTTP status code from the input list and each value is the corresponding dictionary from the ALL_HTTP_CODES dictionary.

Parameters:

Name	Type	Description	Default
codes	list	A list of HTTP status codes.	required
description_only	bool	If True, only the description of the codes will be returned.	False

Returns:

Name	Type	Description
dict	Dict[int, Union[str, Dict[str, str]]]	A dictionary where each key is an HTTP error code from the input
	Dict[int, Union[str, Dict[str, str]]]	list and each value depends on the description_only parameter. If
	Dict[int, Union[str, Dict[str, str]]]	description_only is True, the value is the description string. If
	Dict[int, Union[str, Dict[str, str]]]	description_only is False, the value is a dictionary with keys
	Dict[int, Union[str, Dict[str, str]]]	'description', 'extended_description', and 'link'.

Example:

from dsg_lib.fastapi_functions import http_codes

# Generate a dictionary for HTTP status codes 200 and 404
status_dict = http_codes.generate_code_dict([200, 404])
print(status_dict)
# {200: {'description': 'OK', 'extended_description': 'The request has succeeded', 'link': 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/200'},
#  404: {'description': 'Not Found', 'extended_description': 'The requested resource could not be found', 'link': 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404'}}

# Generate a dictionary for HTTP status codes 200 and 404 with only descriptions
status_dict = http_codes.generate_code_dict([200, 404], description_only=True)
print(status_dict)  # {200: 'OK', 404: 'Not Found'}

Source code in dsg_lib/fastapi_functions/http_codes.py

def generate_code_dict(
    codes: List[int], description_only: bool = False
) -> Dict[int, Union[str, Dict[str, str]]]:
    """
    Generate a dictionary of specific HTTP error codes from the http_codes
    dictionary.

    This function takes a list of HTTP status codes and an optional boolean
    flag. If the flag is True, the function returns a dictionary where each key
    is an HTTP status code from the input list and each value is the
    corresponding description from the ALL_HTTP_CODES dictionary. If the flag is
    False, the function returns a dictionary where each key is an HTTP status
    code from the input list and each value is the corresponding dictionary from
    the ALL_HTTP_CODES dictionary.

    Args:
        codes (list): A list of HTTP status codes.
        description_only (bool, optional): If True, only the description of the codes will be returned.
        Defaults to False.

    Returns:
        dict: A dictionary where each key is an HTTP error code from the input
        list and each value depends on the description_only parameter. If
        description_only is True, the value is the description string. If
        description_only is False, the value is a dictionary with keys
        'description', 'extended_description', and 'link'.

    Example:
    ```python

    from dsg_lib.fastapi_functions import http_codes

    # Generate a dictionary for HTTP status codes 200 and 404
    status_dict = http_codes.generate_code_dict([200, 404])
    print(status_dict)
    # {200: {'description': 'OK', 'extended_description': 'The request has succeeded', 'link': 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/200'},
    #  404: {'description': 'Not Found', 'extended_description': 'The requested resource could not be found', 'link': 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404'}}

    # Generate a dictionary for HTTP status codes 200 and 404 with only descriptions
    status_dict = http_codes.generate_code_dict([200, 404], description_only=True)
    print(status_dict)  # {200: 'OK', 404: 'Not Found'}
    ```
    """

    if description_only:
        # Log the operation
        logger.debug(f"description_only is True and returning HTTP codes: {codes}")

        # If description_only is True, return a dictionary where each key is an
        # HTTP error code from the input list and each value is the
        # corresponding description from the ALL_HTTP_CODES dictionary.
        return {
            code: ALL_HTTP_CODES[code]["description"]
            for code in codes
            if code in ALL_HTTP_CODES
        }
    else:
        # Log the operation
        logger.debug(f"returning HTTP codes: {codes}")

        # If description_only is False, return a dictionary where each key is an
        # HTTP error code from the input list and each value is the
        # corresponding dictionary from the ALL_HTTP_CODES dictionary.
        return {code: ALL_HTTP_CODES[code] for code in codes if code in ALL_HTTP_CODES}

dsg_lib.fastapi_functions._all_codes

This module contains a dictionary mapping HTTP status codes to their descriptions, extended descriptions, and links to their documentation.

Each key in this dictionary is an HTTP status code, and each value is another dictionary with keys 'description', 'extended_description', and 'link'.

The 'description' key maps to a brief string that describes the HTTP status code. The 'extended_description' key maps to a more detailed explanation of the status code. The 'link' key maps to a string that is a link to the documentation for the HTTP status code.

Example

from dsg_lib.fastapi_functions.http_codes import ALL_HTTP_CODES

# Get the dictionary for HTTP status code 200
status_200 = ALL_HTTP_CODES[200]
print(status_200)
# Output: {'description': 'OK', 'extended_description': 'The request has succeeded.', 'link': 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/200'}

# Get the description for HTTP status code 404
description_404 = ALL_HTTP_CODES[404]['description']
print(description_404)  # Output: 'Not Found'

# Get the extended description for HTTP status code 200
extended_description_200 = ALL_HTTP_CODES[200]['extended_description']
print(extended_description_200)  # Output: 'The request has succeeded.'

# Get the link to the documentation for HTTP status code 500
link_500 = ALL_HTTP_CODES[500]['link']
print(link_500)  # Output: 'https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500'

Author: Mike Ryan Date: 2024/05/16 License: MIT

Reference

dsg_lib.fastapi_functions.system_health_endpoints

This module provides a configurable health endpoint for a FastAPI application. It includes the following routes:

• /api/health/status: Returns the status of the application. If the application is running, it will return {"status": "UP"}. This endpoint can be enabled or disabled using the configuration.

• /api/health/uptime: Returns the uptime of the application in a dictionary with the keys "Days", "Hours", "Minutes", and "Seconds". The uptime is calculated from the time the application was started. This endpoint can be enabled or disabled using the configuration.

• /api/health/heapdump: Returns a heap dump of the application. The heap dump is a list of dictionaries, each representing a line of code. Each dictionary includes the filename, line number, size of memory consumed, and the number of times the line is referenced. This endpoint can be enabled or disabled using the configuration.

The module uses the FastAPI, time, tracemalloc, loguru, packaging, and dsg_lib.fastapi.http_codes modules.

Functions:

Name	Description
create_health_router	dict) -> FastAPI.APIRouter: Creates a FastAPI router with health endpoints based on the provided configuration.

Example

from FastAPI import FastAPI
from dsg_lib.fastapi_functions import
system_health_endpoints

app = FastAPI()

# User configuration
config = {
    "enable_status_endpoint": True,
    "enable_uptime_endpoint": False,
    "enable_heapdump_endpoint": True,
}

# Health router
health_router =
system_health_endpoints.create_health_router(config)
app.include_router(health_router, prefix="/api/health",
tags=["system-health"])

# Get the status of the application
response = client.get("/api/health/status")
print(response.json())  # {"status": "UP"}

# Get the uptime of the application response =
client.get("/api/health/uptime")
print(response.json())
# {"uptime": {"Days": 0, "Hours": 0, "Minutes": 1, "Seconds": 42.17}}

# Get the heap dump of the application response =
client.get("/api/health/heapdump")
print(response.json())
# {"memory_use":{"current": "123456", "peak": "789012"}, "heap_dump": [{"filename": "main.py", "lineno": 10, "size": 1234, "count": 1}, ...]}

Author: Mike Ryan Date: 2024/05/16 License: MIT

create_health_router(config)

Create a health router with the following endpoints:

• /status: Returns the status of the application. This endpoint can be enabled or disabled using the enable_status_endpoint key in the configuration.

• /uptime: Returns the uptime of the application. This endpoint can be enabled or disabled using the enable_uptime_endpoint key in the configuration.

• /heapdump: Returns a heap dump of the application. This endpoint can be enabled or disabled using the enable_heapdump_endpoint key in the configuration.

Parameters:

Name	Type	Description	Default
config	dict	A dictionary with the configuration for the endpoints.	required

Returns:

Name	Type	Description
APIRouter		A FastAPI router with the configured endpoints.

Example

from FastAPI import FastAPI
from dsg_lib.fastapi_functions import
system_health_endpoints

app = FastAPI()

# User configuration
config = {
    "enable_status_endpoint": True,
    "enable_uptime_endpoint": False,
    "enable_heapdump_endpoint": True,
}

# Health router
health_router =
system_health_endpoints.create_health_router(config)
app.include_router(health_router, prefix="/api/health",
tags=["system-health"])

# Get the status of the application
response = client.get("/api/health/status")
print(response.json())  # {"status": "UP"}

# Get the uptime of the application response =
client.get("/api/health/uptime")
print(response.json())
# {"uptime": {"Days": 0, "Hours": 0, "Minutes": 1, "Seconds": 42.17}}

# Get the heap dump of the application response =
client.get("/api/health/heapdump")
print(response.json())
# {"memory_use":{"current": "123456", "peak": "789012"}, "heap_dump": [{"filename": "main.py", "lineno": 10, "size": 1234, "count": 1}, ...]}

Source code in dsg_lib/fastapi_functions/system_health_endpoints.py

def create_health_router(config: dict):
    """
    Create a health router with the following endpoints:

    - `/status`: Returns the status of the application. This endpoint can be
      enabled or disabled using the `enable_status_endpoint` key in the
      configuration.

    - `/uptime`: Returns the uptime of the application. This endpoint can be
      enabled or disabled using the `enable_uptime_endpoint` key in the
      configuration.

    - `/heapdump`: Returns a heap dump of the application. This endpoint can be
      enabled or disabled using the `enable_heapdump_endpoint` key in the
      configuration.

    Args:
        config (dict): A dictionary with the configuration for the endpoints.
        Each key should be the name of an endpoint (e.g.,
        `enable_status_endpoint`) and the value should be a boolean indicating
        whether the endpoint is enabled or not.

    Returns:
        APIRouter: A FastAPI router with the configured endpoints.

    Example:
        ```python
        from FastAPI import FastAPI
        from dsg_lib.fastapi_functions import
        system_health_endpoints

        app = FastAPI()

        # User configuration
        config = {
            "enable_status_endpoint": True,
            "enable_uptime_endpoint": False,
            "enable_heapdump_endpoint": True,
        }

        # Health router
        health_router =
        system_health_endpoints.create_health_router(config)
        app.include_router(health_router, prefix="/api/health",
        tags=["system-health"])

        # Get the status of the application
        response = client.get("/api/health/status")
        print(response.json())  # {"status": "UP"}

        # Get the uptime of the application response =
        client.get("/api/health/uptime")
        print(response.json())
        # {"uptime": {"Days": 0, "Hours": 0, "Minutes": 1, "Seconds": 42.17}}

        # Get the heap dump of the application response =
        client.get("/api/health/heapdump")
        print(response.json())
        # {"memory_use":{"current": "123456", "peak": "789012"}, "heap_dump": [{"filename": "main.py", "lineno": 10, "size": 1234, "count": 1}, ...]}

        ```
    """
    # Try to import FastAPI, handle ImportError if FastAPI is not installed
    try:
        import fastapi
        from fastapi import APIRouter, HTTPException, status
        from fastapi.responses import ORJSONResponse
    except ImportError:  # pragma: no cover
        APIRouter = HTTPException = status = ORJSONResponse = fastapi = (
            None  # pragma: no cover
        )

    # Check FastAPI version
    min_version = "0.100.0"  # replace with your minimum required version
    if fastapi is not None and packaging_version.parse(
        fastapi.__version__
    ) < packaging_version.parse(min_version):
        raise ImportError(
            f"FastAPI version >= {min_version} required, run `pip install --upgrade fastapi`"
        )  # pragma: no cover

    # Store the start time of the application
    app_start_time = time.time()

    # TODO: determine method to shutdown/restart python application

    status_response = generate_code_dict([400, 405, 500], description_only=False)

    tracemalloc.start()
    # Create a new router
    router = APIRouter()

    # Check if the status endpoint is enabled in the configuration
    if config.get("enable_status_endpoint", True):
        # Define the status endpoint
        @router.get(
            "/status",
            tags=["system-health"],
            status_code=status.HTTP_200_OK,
            response_class=ORJSONResponse,
            responses=status_response,
        )
        async def health_status():
            """
            Returns the status of the application.

            This endpoint returns a dictionary with the status of the
            application. If the application is running, it will return
            `{"status": "UP"}`.

            Returns:
                dict: A dictionary with the status of the application. The
                dictionary has a single key, "status", and the value is a string
                that indicates the status of the application.

            Raises:
                HTTPException: If there is an error while getting the status of
                the application.

            Example:
                ```python
                from FastAPI import FastAPI
                from dsg_lib.fastapi_functions import
                system_health_endpoints

                app = FastAPI()

                # User configuration
                config = {
                    "enable_status_endpoint": True,
                    "enable_uptime_endpoint": False,
                    "enable_heapdump_endpoint": True,
                }

                # Health router
                health_router =
                system_health_endpoints.create_health_router(config)
                app.include_router(health_router, prefix="/api/health",
                tags=["system-health"])

                # Get the status of the application
                response = client.get("/api/health/status")
                print(response.json())  # {"status": "UP"}
            ```
            """
            # Log the status request
            logger.info("Health status of up returned")
            # Return a dictionary with the status of the application
            return {"status": "UP"}

    # Check if the uptime endpoint is enabled in the configuration
    if config.get("enable_uptime_endpoint", True):
        # Define the uptime endpoint
        @router.get("/uptime", response_class=ORJSONResponse, responses=status_response)
        async def get_uptime():
            """
            Calculate and return the uptime of the application.

            This endpoint returns a dictionary with the uptime of the
            application. The uptime is calculated from the time the application
            was started and is returned in a dictionary with the keys "Days",
            "Hours", "Minutes", and "Seconds".

            Returns:
                dict: A dictionary with the uptime of the application. The
                dictionary has keys for "Days", "Hours", "Minutes", and
                "Seconds".

            Raises:
                HTTPException: If there is an error while calculating the uptime
                of the application.

            Example:
                ```python
                from FastAPI import FastAPI
                from dsg_lib.fastapi_functions import
                system_health_endpoints

                app = FastAPI()

                # User configuration
                config = {
                    "enable_status_endpoint": True,
                    "enable_uptime_endpoint": False,
                    "enable_heapdump_endpoint": True,
                }

                # Health router
                health_router =
                system_health_endpoints.create_health_router(config)
                app.include_router(health_router, prefix="/api/health",
                tags=["system-health"])

                # Get the uptime of the application response =
                client.get("/api/health/uptime")
                print(response.json())
                # {"uptime": {"Days": 0, "Hours": 0, "Minutes": 1, "Seconds": 42.17}}

                ```
            """
            # Calculate the total uptime in seconds This is done by subtracting
            # the time when the application started from the current time
            uptime_seconds = time.time() - app_start_time

            # Convert the uptime from seconds to days, hours, minutes, and
            # seconds
            days, rem = divmod(uptime_seconds, 86400)
            hours, rem = divmod(rem, 3600)
            minutes, seconds = divmod(rem, 60)

            # Log the uptime
            logger.info(
                f"Uptime: {int(days)} days, {int(hours)} hours, {int(minutes)} minutes, {round(seconds, 2)} seconds"
            )

            # Return a dictionary with the uptime The dictionary has keys for
            # days, hours, minutes, and seconds
            return {
                "uptime": {
                    "Days": int(days),
                    "Hours": int(hours),
                    "Minutes": int(minutes),
                    "Seconds": round(seconds, 2),
                }
            }

    if config.get("enable_heapdump_endpoint", True):

        @router.get(
            "/heapdump", response_class=ORJSONResponse, responses=status_response
        )
        async def get_heapdump():
            """
            Returns a heap dump of the application.

            This endpoint returns a snapshot of the current memory usage of the
            application. The heap dump is a list of dictionaries, each
            representing a line of code. Each dictionary includes the filename,
            line number, size of memory consumed, and the number of times the
            line is referenced.

            Returns:
                dict: A dictionary with the current and peak memory usage, and
                the heap dump of the application. The dictionary has two keys,
                "memory_use" and "heap_dump". The "memory_use" key contains a
                dictionary with the current and peak memory usage. The
                "heap_dump" key contains a list of dictionaries, each
                representing a line of code.

            Raises:
                HTTPException: If there is an error while getting the heap dump
                of the application.

            Example:
                ```python
                from FastAPI import FastAPI
                from dsg_lib.fastapi_functions import
                system_health_endpoints

                app = FastAPI()

                # User configuration
                config = {
                    "enable_status_endpoint": True,
                    "enable_uptime_endpoint": False,
                    "enable_heapdump_endpoint": True,
                }

                # Health router
                health_router =
                system_health_endpoints.create_health_router(config)
                app.include_router(health_router, prefix="/api/health",
                tags=["system-health"])

                # Get the heap dump of the application response =
                client.get("/api/health/heapdump")
                print(response.json())
                # {"memory_use":{"current": "123456", "peak": "789012"}, "heap_dump": [{"filename": "main.py", "lineno": 10, "size": 1234, "count": 1}, ...]}

                ```
            """

            try:
                # Take a snapshot of the current memory usage
                snapshot = tracemalloc.take_snapshot()
                # Get the top 10 lines consuming memory
                top_stats = snapshot.statistics("traceback")

                heap_dump = []
                for stat in top_stats[:10]:
                    # Get the first frame from the traceback
                    frame = stat.traceback[0]
                    # Add the frame to the heap dump
                    heap_dump.append(
                        {
                            "filename": frame.filename,
                            "lineno": frame.lineno,
                            "size": stat.size,
                            "count": stat.count,
                        }
                    )

                logger.info(f"Heap dump returned {heap_dump}")
                memory_use = tracemalloc.get_traced_memory()
                return {
                    "memory_use": {
                        "current": f"{memory_use[0]:,}",
                        "peak": f"{memory_use[1]:,}",
                    },
                    "heap_dump": heap_dump,
                }
            except Exception as ex:
                logger.error(f"Error in get_heapdump: {ex}")
                raise HTTPException(
                    status_code=500, detail=f"Error in get_heapdump: {ex}"
                )

    return router

Ended: FastAPI Functions

Database Functions

Reference

dsg_lib.async_database_functions.base_schema

This module defines the base schema for database models in the application.

The module uses SQLAlchemy as the ORM and provides a SchemaBase class that all other models should inherit from. The SchemaBase class includes common columns that are needed for most models like pkid, date_created, and date_updated.

• pkid: A unique identifier for each record. It's a string representation of a UUID.
• date_created: The date and time when a particular row was inserted into the table. It defaults to the current UTC time when the instance is created.
• date_updated: The date and time when a particular row was last updated. It defaults to the current UTC time whenever the instance is updated.

To create a new database model, import this module and extend the SchemaBase class.

Example:

from dsg_lib.async_database_functions import base_schema

class MyModel(base_schema.SchemaBaseSQLite):
        # Define your model-specific columns here my_column =
        base_schema.Column(base_schema.String(50))

Author: Mike Ryan Date: 2024/05/16 License: MIT

SchemaBaseCockroachDB

This class provides a base schema that includes common columns for most models when using a CockroachDB database. CockroachDB uses the same syntax as PostgreSQL. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBaseCockroachDB, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseCockroachDB:
    """
    This class provides a base schema that includes common columns for most
    models when using a CockroachDB database. CockroachDB uses the same syntax
    as PostgreSQL. All other models should inherit from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBaseCockroachDB, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')"),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')"),
        onupdate=text("(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')"),
        comment=date_updated_comment,
    )

SchemaBaseFirebird

This class provides a base schema that includes common columns for most models when using a Firebird database. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBaseFirebird, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseFirebird:
    """
    This class provides a base schema that includes common columns for most
    models when using a Firebird database. All other models should inherit
    from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBaseFirebird, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("CURRENT_TIMESTAMP"),
        comment="Date and time when a row was inserted, defaults to current time",
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("CURRENT_TIMESTAMP"),
        onupdate=text("CURRENT_TIMESTAMP"),
        comment="Date and time when a row was last updated, defaults to current time on update",
    )

SchemaBaseMSSQL

This class provides a base schema that includes common columns for most models when using a Microsoft SQL Server database. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBaseMSSQL, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseMSSQL:
    """
    This class provides a base schema that includes common columns for most
    models when using a Microsoft SQL Server database. All other models should
    inherit from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBaseMSSQL, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("GETUTCDATE()"),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("GETUTCDATE()"),
        onupdate=text("GETUTCDATE()"),
        comment=date_updated_comment,
    )

SchemaBaseMySQL

This class provides a base schema that includes common columns for most models when using a MySQL database. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBaseMySQL, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseMySQL:
    """
    This class provides a base schema that includes common columns for most
    models when using a MySQL database. All other models should inherit
    from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBaseMySQL, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("UTC_TIMESTAMP()"),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("UTC_TIMESTAMP()"),
        onupdate=text("UTC_TIMESTAMP()"),
        comment=date_updated_comment,
    )

SchemaBaseOracle

This class provides a base schema that includes common columns for most models when using an Oracle database. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBaseOracle, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseOracle:
    """
    This class provides a base schema that includes common columns for most
    models when using an Oracle database. All other models should inherit
    from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBaseOracle, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("SYS_EXTRACT_UTC(SYSTIMESTAMP)"),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("SYS_EXTRACT_UTC(SYSTIMESTAMP)"),
        onupdate=text("SYS_EXTRACT_UTC(SYSTIMESTAMP)"),
        comment=date_updated_comment,
    )

SchemaBasePostgres

This class provides a base schema that includes common columns for most models when using a PostgreSQL database. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBasePostgres, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBasePostgres:
    """
    This class provides a base schema that includes common columns for most
    models when using a PostgreSQL database. All other models should inherit
    from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBasePostgres, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')"),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')"),
        onupdate=text("(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')"),
        comment=date_updated_comment,
    )

SchemaBaseSQLite

This class provides a base schema that includes common columns for most models. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBase, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseSQLite:
    """
    This class provides a base schema that includes common columns for most
    models. All other models should inherit from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBase, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        default=datetime.datetime.now(datetime.timezone.utc),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        default=datetime.datetime.now(datetime.timezone.utc),
        onupdate=datetime.datetime.now(datetime.timezone.utc),
        comment=date_updated_comment,
    )

SchemaBaseSybase

This class provides a base schema that includes common columns for most models when using a Sybase database. All other models should inherit from this class.

Attributes:

Name	Type	Description
pkid	str	A unique identifier for each record. It's a string
date_created	datetime	The date and time when a particular row was
date_updated	datetime	The date and time when a particular row was

Example:

from dsg_lib.async_database_functions import base_schema
from sqlalchemy.orm import declarative_base

BASE = declarative_base()

class MyModel(base_schema.SchemaBaseSybase, BASE):
    # Define your model-specific columns here
    my_column = base_schema.Column(base_schema.String(50))

Source code in dsg_lib/async_database_functions/base_schema.py

class SchemaBaseSybase:
    """
    This class provides a base schema that includes common columns for most
    models when using a Sybase database. All other models should inherit
    from this class.

    Attributes:
        pkid (str): A unique identifier for each record. It's a string
        representation of a UUID.
        date_created (datetime): The date and time when a particular row was
        inserted into the table. It defaults to the current UTC time when the
        instance is created.
        date_updated (datetime): The date and time when a particular row was
        last updated. It defaults to the current UTC time whenever the instance
        is updated.

    Example:
    ```python
    from dsg_lib.async_database_functions import base_schema
    from sqlalchemy.orm import declarative_base

    BASE = declarative_base()

    class MyModel(base_schema.SchemaBaseSybase, BASE):
        # Define your model-specific columns here
        my_column = base_schema.Column(base_schema.String(50))
    ```
    """

    # Each instance in the table will have a unique id which is a string
    # representation of a UUID
    pkid = Column(
        String(36),
        primary_key=True,
        index=True,
        default=lambda: str(uuid4()),
        comment=uuid_comment,
    )

    # The date and time when a particular row was inserted into the table. It
    # defaults to the current UTC time when the instance is created.
    date_created = Column(
        DateTime,
        index=True,
        server_default=text("GETUTCDATE()"),
        comment=date_created_comment,
    )

    # The date and time when a particular row was last updated. It defaults to
    # the current UTC time whenever the instance is updated.
    date_updated = Column(
        DateTime,
        index=True,
        server_default=text("GETUTCDATE()"),
        onupdate=text("GETUTCDATE()"),
        comment=date_updated_comment,
    )

Reference

dsg_lib.async_database_functions.database_config

This module provides classes and functions for managing asynchronous database operations using SQLAlchemy and asyncio.

The main classes are DBConfig, which manages the database configuration and creates a SQLAlchemy engine and a MetaData instance, and AsyncDatabase, which uses an instance of DBConfig to perform asynchronous database operations.

The module also provides a function, import_sqlalchemy, which tries to import SQLAlchemy and its components, and raises an ImportError if SQLAlchemy is not installed or if the installed version is not compatible.

The module uses the logger from the dsg_lib for logging, and the time module for working with times. It also uses the contextlib module for creating context managers, and the typing module for type hinting.

The BASE variable is a base class for declarative database models. It is created using the declarative_base function from sqlalchemy.orm.

This module is part of the dsg_lib package, which provides utilities for working with databases in Python.

Example:

from dsg_lib.async_database_functions import database_config

# Define your database configuration config = {
    "database_uri": "postgresql+asyncpg://user:password@localhost/dbname",
    "echo": True, "future": True, "pool_pre_ping": True, "pool_size": 5,
    "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
}

# Create a DBConfig instance db_config = database_config.DBConfig(config)

# Use the DBConfig instance to get a database session async with
db_config.get_db_session() as session:
    # Perform your database operations here pass

Author: Mike Ryan Date: 2024/05/16 License: MIT

DBConfig

A class used to manage the database configuration and create a SQLAlchemy engine.

Attributes:

Name	Type	Description
config	dict	A dictionary containing the database configuration
parameters.	engine (Engine	The SQLAlchemy engine created with the
database	URI from the config. metadata (MetaData	The SQLAlchemy

Create Engine Support Functions by Database Type Confirmed by testing [SQLITE, PostrgeSQL] To Be Tested [MySQL, Oracle, MSSQL] 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

Example:

from dsg_lib.async_database_functions import database_config

# Define your database configuration config = {
    "database_uri": "postgresql+asyncpg://user:password@localhost/dbname",
    "echo": True, "future": True, "pool_pre_ping": True, "pool_size": 5,
    "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
}

# Create a DBConfig instance db_config = database_config.DBConfig(config)

# Use the DBConfig instance to get a database session async with
db_config.get_db_session() as session:
    # Perform your database operations here pass

Source code in dsg_lib/async_database_functions/database_config.py

class DBConfig:
    """
    A class used to manage the database configuration and create a SQLAlchemy
    engine.

    Attributes:
        config (dict): A dictionary containing the database configuration
        parameters. engine (Engine): The SQLAlchemy engine created with the
        database URI from the config. metadata (MetaData): The SQLAlchemy
        MetaData instance.


    Create Engine Support Functions by Database Type Confirmed by testing
    [SQLITE, PostrgeSQL] To Be Tested [MySQL, Oracle, MSSQL] 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

    Example:
    ```python

    from dsg_lib.async_database_functions import database_config

    # Define your database configuration config = {
        "database_uri": "postgresql+asyncpg://user:password@localhost/dbname",
        "echo": True, "future": True, "pool_pre_ping": True, "pool_size": 5,
        "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
    }

    # Create a DBConfig instance db_config = database_config.DBConfig(config)

    # Use the DBConfig instance to get a database session async with
    db_config.get_db_session() as session:
        # Perform your database operations here pass
    ```

    """

    SUPPORTED_PARAMETERS = {
        "sqlite": {"echo", "future", "pool_recycle"},
        "postgresql": {
            "echo",
            "future",
            "pool_pre_ping",
            "pool_size",
            "max_overflow",
            "pool_recycle",
            "pool_timeout",
        },
        # Add other engines here...
    }

    def __init__(self, config: Dict):
        """
        Initializes the DBConfig instance with the given database configuration.

        The configuration should be a dictionary with the following keys: -
        "database_uri": The URI for the database. - "echo": If True, the engine
        will log all statements as well as a `repr()` of their parameter lists
        to the engines logger, which defaults to sys.stdout. - "future": If
        True, use the future version of SQLAlchemy, which supports asyncio. -
        "pool_pre_ping": If True, the pool will test the connection for liveness
        upon each checkout. - "pool_size": The size of the connection pool to be
        maintained. - "max_overflow": The number of connections that can be
        opened above the `pool_size` setting, when all other connections are in
        use. - "pool_recycle": The number of seconds after which a connection is
        automatically recycled. This is required for MySQL, which removes
        connections after 8 hours idle by default. - "pool_timeout": The number
        of seconds to wait before giving up on getting a connection from the
        pool.

        Args:
            config (Dict): A dictionary containing the database configuration
            parameters.

        Raises:
            Exception: If there are unsupported parameters for the database
            engine type.

        Example:
        ```python

        from dsg_lib.async_database_functions import database_config

        # Define your database configuration config = {
            "database_uri":
            "postgresql+asyncpg://user:password@localhost/dbname", "echo": True,
            "future": True, "pool_pre_ping": True, "pool_size": 5,
            "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
        }

        # Create a DBConfig instance db_config =
        database_config.DBConfig(config)
        ```
        """
        self.config = config
        engine_type = self.config["database_uri"].split("+")[0]
        supported_parameters = self.SUPPORTED_PARAMETERS.get(engine_type, set())
        unsupported_parameters = (
            set(config.keys()) - supported_parameters - {"database_uri"}
        )
        if unsupported_parameters:
            error_message = (
                f"Unsupported parameters for {engine_type}: {unsupported_parameters}"
            )
            logger.error(error_message)
            raise Exception(error_message)

        engine_parameters = {
            param: self.config.get(param)
            for param in supported_parameters
            if self.config.get(param) is not None
        }
        self.engine = create_async_engine(
            self.config["database_uri"], **engine_parameters
        )
        self.metadata = MetaData()

    @asynccontextmanager
    async def get_db_session(self):
        """
        This method returns a context manager that provides a new database
        session.

        The session is created using the SQLAlchemy engine from the DBConfig
        instance, and it does not expire on commit. The session is of type
        AsyncSession.

        This method should be used with the `async with` statement.

        Yields:
            AsyncSession: A new SQLAlchemy asynchronous session.

        Raises:
            SQLAlchemyError: If a database error occurs.

        Example:
        ```python

        from dsg_lib.async_database_functions import database_config

        # Define your database configuration config = {
            "database_uri":
            "postgresql+asyncpg://user:password@localhost/dbname", "echo": True,
            "future": True, "pool_pre_ping": True, "pool_size": 5,
            "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
        }

        # Create a DBConfig instance db_config =
        database_config.DBConfig(config)

        # Use the DBConfig instance to get a database session async with
        db_config.get_db_session() as session:
            # Perform your database operations here pass
        ```
        """
        logger.debug("Creating new database session")
        try:
            # Create a new database session
            async with sessionmaker(
                self.engine, expire_on_commit=False, class_=AsyncSession
            )() as session:
                # Yield the session to the context manager
                yield session
        except SQLAlchemyError as e:  # pragma: no cover
            # Log the error and raise it
            logger.error(f"Database error occurred: {str(e)}")  # pragma: no cover
            raise  # pragma: no cover
        finally:  # pragma: no cover
            # Log the end of the database session
            logger.debug("Database session ended")  # pragma: no cover

__init__(config)

Initializes the DBConfig instance with the given database configuration.

The configuration should be a dictionary with the following keys: - "database_uri": The URI for the database. - "echo": If True, the engine will log all statements as well as a repr() of their parameter lists to the engines logger, which defaults to sys.stdout. - "future": If True, use the future version of SQLAlchemy, which supports asyncio. - "pool_pre_ping": If True, the pool will test the connection for liveness upon each checkout. - "pool_size": The size of the connection pool to be maintained. - "max_overflow": The number of connections that can be opened above the pool_size setting, when all other connections are in use. - "pool_recycle": The number of seconds after which a connection is automatically recycled. This is required for MySQL, which removes connections after 8 hours idle by default. - "pool_timeout": The number of seconds to wait before giving up on getting a connection from the pool.

Parameters:

Name	Type	Description	Default
config	Dict	A dictionary containing the database configuration	required

Raises:

Type	Description
Exception	If there are unsupported parameters for the database

Example:

from dsg_lib.async_database_functions import database_config

# Define your database configuration config = {
    "database_uri":
    "postgresql+asyncpg://user:password@localhost/dbname", "echo": True,
    "future": True, "pool_pre_ping": True, "pool_size": 5,
    "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
}

# Create a DBConfig instance db_config =
database_config.DBConfig(config)

Source code in dsg_lib/async_database_functions/database_config.py

def __init__(self, config: Dict):
    """
    Initializes the DBConfig instance with the given database configuration.

    The configuration should be a dictionary with the following keys: -
    "database_uri": The URI for the database. - "echo": If True, the engine
    will log all statements as well as a `repr()` of their parameter lists
    to the engines logger, which defaults to sys.stdout. - "future": If
    True, use the future version of SQLAlchemy, which supports asyncio. -
    "pool_pre_ping": If True, the pool will test the connection for liveness
    upon each checkout. - "pool_size": The size of the connection pool to be
    maintained. - "max_overflow": The number of connections that can be
    opened above the `pool_size` setting, when all other connections are in
    use. - "pool_recycle": The number of seconds after which a connection is
    automatically recycled. This is required for MySQL, which removes
    connections after 8 hours idle by default. - "pool_timeout": The number
    of seconds to wait before giving up on getting a connection from the
    pool.

    Args:
        config (Dict): A dictionary containing the database configuration
        parameters.

    Raises:
        Exception: If there are unsupported parameters for the database
        engine type.

    Example:
    ```python

    from dsg_lib.async_database_functions import database_config

    # Define your database configuration config = {
        "database_uri":
        "postgresql+asyncpg://user:password@localhost/dbname", "echo": True,
        "future": True, "pool_pre_ping": True, "pool_size": 5,
        "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
    }

    # Create a DBConfig instance db_config =
    database_config.DBConfig(config)
    ```
    """
    self.config = config
    engine_type = self.config["database_uri"].split("+")[0]
    supported_parameters = self.SUPPORTED_PARAMETERS.get(engine_type, set())
    unsupported_parameters = (
        set(config.keys()) - supported_parameters - {"database_uri"}
    )
    if unsupported_parameters:
        error_message = (
            f"Unsupported parameters for {engine_type}: {unsupported_parameters}"
        )
        logger.error(error_message)
        raise Exception(error_message)

    engine_parameters = {
        param: self.config.get(param)
        for param in supported_parameters
        if self.config.get(param) is not None
    }
    self.engine = create_async_engine(
        self.config["database_uri"], **engine_parameters
    )
    self.metadata = MetaData()

get_db_session() async

This method returns a context manager that provides a new database session.

The session is created using the SQLAlchemy engine from the DBConfig instance, and it does not expire on commit. The session is of type AsyncSession.

This method should be used with the async with statement.

Yields:

Name	Type	Description
AsyncSession		A new SQLAlchemy asynchronous session.

Raises:

Type	Description
SQLAlchemyError	If a database error occurs.

Example:

from dsg_lib.async_database_functions import database_config

# Define your database configuration config = {
    "database_uri":
    "postgresql+asyncpg://user:password@localhost/dbname", "echo": True,
    "future": True, "pool_pre_ping": True, "pool_size": 5,
    "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
}

# Create a DBConfig instance db_config =
database_config.DBConfig(config)

# Use the DBConfig instance to get a database session async with
db_config.get_db_session() as session:
    # Perform your database operations here pass

Source code in dsg_lib/async_database_functions/database_config.py

@asynccontextmanager
async def get_db_session(self):
    """
    This method returns a context manager that provides a new database
    session.

    The session is created using the SQLAlchemy engine from the DBConfig
    instance, and it does not expire on commit. The session is of type
    AsyncSession.

    This method should be used with the `async with` statement.

    Yields:
        AsyncSession: A new SQLAlchemy asynchronous session.

    Raises:
        SQLAlchemyError: If a database error occurs.

    Example:
    ```python

    from dsg_lib.async_database_functions import database_config

    # Define your database configuration config = {
        "database_uri":
        "postgresql+asyncpg://user:password@localhost/dbname", "echo": True,
        "future": True, "pool_pre_ping": True, "pool_size": 5,
        "max_overflow": 10, "pool_recycle": 3600, "pool_timeout": 30,
    }

    # Create a DBConfig instance db_config =
    database_config.DBConfig(config)

    # Use the DBConfig instance to get a database session async with
    db_config.get_db_session() as session:
        # Perform your database operations here pass
    ```
    """
    logger.debug("Creating new database session")
    try:
        # Create a new database session
        async with sessionmaker(
            self.engine, expire_on_commit=False, class_=AsyncSession
        )() as session:
            # Yield the session to the context manager
            yield session
    except SQLAlchemyError as e:  # pragma: no cover
        # Log the error and raise it
        logger.error(f"Database error occurred: {str(e)}")  # pragma: no cover
        raise  # pragma: no cover
    finally:  # pragma: no cover
        # Log the end of the database session
        logger.debug("Database session ended")  # pragma: no cover

Reference

dsg_lib.async_database_functions.async_database

async_database.py.

This module provides classes for managing asynchronous database operations using SQLAlchemy and asyncio.

Classes:

Name	Description
- DBConfig	Manages the database configuration.
- AsyncDatabase	Manages the asynchronous database operations.

The DBConfig class initializes the database configuration and creates a SQLAlchemy engine and a MetaData instance.

The AsyncDatabase class uses an instance of DBConfig to perform asynchronous database operations. It provides methods to get a database session and to create tables in the database.

This module uses the logger from the dsg_lib.common_functions for logging.

Example:

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)

Author: Mike Ryan Date: 2024/05/16 License: MIT

AsyncDatabase

A class used to manage the asynchronous database operations.

Attributes

db_config : DBConfig an instance of DBConfig class containing the database configuration Base : Base the declarative base model for SQLAlchemy

Methods

get_db_session(): Returns a context manager that provides a new database session. create_tables(): Asynchronously creates all tables in the database.

Source code in dsg_lib/async_database_functions/async_database.py

class AsyncDatabase:
    """
    A class used to manage the asynchronous database operations.

    Attributes
    ----------
    db_config : DBConfig
        an instance of DBConfig class containing the database configuration
    Base : Base
        the declarative base model for SQLAlchemy

    Methods
    -------
    get_db_session():
        Returns a context manager that provides a new database session.
    create_tables():
        Asynchronously creates all tables in the database.
    """

    def __init__(self, db_config: DBConfig):
        """Initialize the AsyncDatabase class with an instance of DBConfig.

        Parameters:
        db_config (DBConfig): An instance of DBConfig class containing the
        database configuration.

        Returns: None
        """
        self.db_config = db_config
        self.Base = BASE
        logger.debug("AsyncDatabase initialized")

    def get_db_session(self):
        """This method returns a context manager that provides a new database
        session.

        Parameters: None

        Returns: contextlib._GeneratorContextManager: A context manager that
        provides a new database session.
        """
        logger.debug("Getting database session")
        return self.db_config.get_db_session()

    async def create_tables(self):
        """This method asynchronously creates all tables in the database.

        Parameters: None

        Returns: None
        """
        logger.debug("Creating tables")
        try:
            # Bind the engine to the metadata of the base class
            self.Base.metadata.bind = self.db_config.engine

            # Begin a new transaction
            async with self.db_config.engine.begin() as conn:
                # Run a function in a synchronous manner
                await conn.run_sync(self.Base.metadata.create_all)
            logger.info("Tables created successfully")
        except Exception as ex:  # pragma: no cover
            # Log the error and raise it
            logger.error(f"Error creating tables: {ex}")  # pragma: no cover
            raise  # pragma: no cover

    async def disconnect(self):  # pragma: no cover
        """
        This method asynchronously disconnects the database engine.

        Parameters: None

        Returns: None
        """
        logger.debug("Disconnecting from database")
        try:
            await self.db_config.engine.dispose()
            logger.info("Disconnected from database")
        except Exception as ex:  # pragma: no cover
            logger.error(f"Error disconnecting from database: {ex}")  # pragma: no cover
            raise  # pragma: no cover

__init__(db_config)

Initialize the AsyncDatabase class with an instance of DBConfig.

Parameters: db_config (DBConfig): An instance of DBConfig class containing the database configuration.

Returns: None

Source code in dsg_lib/async_database_functions/async_database.py

def __init__(self, db_config: DBConfig):
    """Initialize the AsyncDatabase class with an instance of DBConfig.

    Parameters:
    db_config (DBConfig): An instance of DBConfig class containing the
    database configuration.

    Returns: None
    """
    self.db_config = db_config
    self.Base = BASE
    logger.debug("AsyncDatabase initialized")

create_tables() async

This method asynchronously creates all tables in the database.

Parameters: None

Returns: None

Source code in dsg_lib/async_database_functions/async_database.py

async def create_tables(self):
    """This method asynchronously creates all tables in the database.

    Parameters: None

    Returns: None
    """
    logger.debug("Creating tables")
    try:
        # Bind the engine to the metadata of the base class
        self.Base.metadata.bind = self.db_config.engine

        # Begin a new transaction
        async with self.db_config.engine.begin() as conn:
            # Run a function in a synchronous manner
            await conn.run_sync(self.Base.metadata.create_all)
        logger.info("Tables created successfully")
    except Exception as ex:  # pragma: no cover
        # Log the error and raise it
        logger.error(f"Error creating tables: {ex}")  # pragma: no cover
        raise  # pragma: no cover

disconnect() async

This method asynchronously disconnects the database engine.

Parameters: None

Returns: None

Source code in dsg_lib/async_database_functions/async_database.py

async def disconnect(self):  # pragma: no cover
    """
    This method asynchronously disconnects the database engine.

    Parameters: None

    Returns: None
    """
    logger.debug("Disconnecting from database")
    try:
        await self.db_config.engine.dispose()
        logger.info("Disconnected from database")
    except Exception as ex:  # pragma: no cover
        logger.error(f"Error disconnecting from database: {ex}")  # pragma: no cover
        raise  # pragma: no cover

get_db_session()

This method returns a context manager that provides a new database session.

Parameters: None

Returns: contextlib._GeneratorContextManager: A context manager that provides a new database session.

Source code in dsg_lib/async_database_functions/async_database.py

def get_db_session(self):
    """This method returns a context manager that provides a new database
    session.

    Parameters: None

    Returns: contextlib._GeneratorContextManager: A context manager that
    provides a new database session.
    """
    logger.debug("Getting database session")
    return self.db_config.get_db_session()

Reference

Configuration Matrix

Create Engine Support Functions by Database Type Confirmed by testing [SQLITE, PostgreSQL] To Be Tested [MySQL, Oracle, MSSQL] 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

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.
- `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.
- `update_many`: [Deprecated] Use `execute_many` with UPDATE queries 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

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.
• 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.
    - `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 (module_name.AsyncDatabase): An instance of the
            AsyncDatabase class for performing asynchronous database operations.

        Example:
        ```python
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )

        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }

        db_config = database_config.DBConfig(config)

        async_db = async_database.AsyncDatabase(db_config)

        db_ops = database_operations.DatabaseOperations(async_db)

        ```
        """
        # 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
        from sqlalchemy import Table, MetaData, Column,
        Integer, String from dsg_lib.async_database_functions import module_name metadata = MetaData()
        my_table = Table('my_table', metadata,
                        Column('id', Integer, primary_key=True), Column('name',
                        String))

        from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)
        # get columns details
        columns = await db_ops.get_columns_details(my_table)
        ```
        """
        # 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
            from sqlalchemy import Table, MetaData, Column, Integer,
                String from dsg_lib.async_database_functions import module_name metadata = MetaData()
                my_table = Table('my_table', metadata,
                                Column('id', Integer, primary_key=True),
                                Column('name', String, primary_key=True))
            from dsg_lib.async_database_functions import (
                async_database,
                base_schema,
                database_config,
                database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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)

            # get primary keys
            primary_keys = await db_ops.get_primary_keys(my_table)
            ```
        """
        # 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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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)
            # get table names
            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

    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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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)
            # count query
            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 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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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)
            # read one record
            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):
        """
        Executes a fetch query on the database and returns a list of records
        that match the query.

        This asynchronous method accepts a SQLAlchemy `Select` query object.
        It returns a list of records that match the query.

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

        Returns:
            list: A list of records that match the query.

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

        Example:
            ```python
            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)
            # read query
            records = await db_ops.read_query(select(User).where(User.age > 30))
            ```
        """
        # 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 = result.scalars().all()
                logger.debug(f"read_query result: {records}")
                # Log the successful query execution
                if all(
                    isinstance(record, tuple) for record in records
                ):  # pragma: no cover
                    logger.debug(f"read_query result is a tuple {type(records)}")
                    # If all records are tuples, convert them to dictionaries
                    records_data = [
                        dict(zip(("request_group_id", "count"), record, strict=False))
                        for record in records
                    ]
                else:
                    logger.debug(f"read_query result is a dictionary {type(records)}")
                    # Otherwise, try to convert the records to dictionaries using the __dict__ attribute
                    records_data = [record.__dict__ for record in records]

                logger.debug(
                    f"Fetch query executed successfully. Records: {records_data}"
                )

                return records

        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 read_multi_query(self, queries: Dict[str, str]):
        """
        Executes multiple fetch queries on the database and returns a dictionary
        of results for each query.

        This asynchronous method takes a dictionary where each key is a query
        name and each value is a SQLAlchemy `Select` query object. The method executes each
        query and returns a dictionary where each key is the query name, and the
        corresponding value is a list of records that match that query.

        Parameters:
            queries (Dict[str, Select]): A dictionary of SQLAlchemy `Select`
            query objects.

        Returns:
            dict: A dictionary where each key is a query name and each value is
            a list of records that match the query.

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

        Example:
            ```python
            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)
            # read multi query
            queries = {
                "query1": select(User).where(User.age > 30),
                "query2": select(User).where(User.age < 20),
            }
            results = await db_ops.read_multi_query(queries)
            ```
        """
        # Log the start of the operation
        logger.debug("Starting read_multi_query operation")

        try:
            results = {}
            # Start a new database session
            async with self.async_db.get_db_session() as session:
                for query_name, query in queries.items():
                    # 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)
                    data = result.scalars().all()

                    # Convert the records to dictionaries for logging
                    data_dicts = [record.__dict__ for record in data]
                    logger.debug(f"Fetch result for query '{query_name}': {data_dicts}")

                    # Store the records in the results dictionary
                    results[query_name] = data
            return results

        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 execute_one(
        self, query: ClauseElement, values: Optional[Dict[str, Any]] = None
    ) -> Union[str, Dict[str, str]]:
        """
        Executes a single non-read SQL query asynchronously.

        This method executes a single SQL statement that modifies the database,
        such as INSERT, UPDATE, or DELETE. It handles the execution within an
        asynchronous session and commits the transaction upon success.

        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.

        Returns:
            Union[str, Dict[str, str]]: "complete" if the query executed and committed successfully,
            or an error dictionary if an exception occurred.

        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}")
                await session.execute(query, params=values)
                await session.commit()
                logger.debug("Query executed successfully")
                return "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]]]]
    ) -> Union[str, Dict[str, str]]:
        """
        Executes multiple non-read SQL queries asynchronously within a single transaction.

        This method executes a list of SQL statements that modify the database,
        such as multiple INSERTs, UPDATEs, or DELETEs. All queries are executed
        within the same transaction, which is committed if all succeed, or rolled
        back if any fail.

        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:
            Union[str, Dict[str, str]]: "complete" if all queries executed and committed successfully,
            or an error dictionary if an exception occurred.

        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:
            async with self.async_db.get_db_session() as session:
                for query, values in queries:
                    logger.debug(f"Executing query: {query}")
                    await session.execute(query, params=values)
                await session.commit()
                logger.debug("All queries executed successfully")
                return "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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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
            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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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 many records
            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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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)
            # update one record
            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 {
                        "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
            from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
            )
            # Create a DBConfig instance
            config = {
                # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
                "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
                "echo": False,
                "future": True,
                # "pool_pre_ping": True,
                # "pool_size": 10,
                # "max_overflow": 10,
                "pool_recycle": 3600,
                # "pool_timeout": 30,
            }
            # 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)
            # delete one record
            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 {
                        "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
        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)
        # Delete multiple records
        deleted_count = await db_ops.delete_many(User, 'id', [1, 2, 3])
        print(f"Deleted {deleted_count} records.")
        ```
        """
        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	required

Example:

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

config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}

db_config = database_config.DBConfig(config)

async_db = async_database.AsyncDatabase(db_config)

db_ops = database_operations.DatabaseOperations(async_db)

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 (module_name.AsyncDatabase): An instance of the
        AsyncDatabase class for performing asynchronous database operations.

    Example:
    ```python
    from dsg_lib.async_database_functions import (
    async_database,
    base_schema,
    database_config,
    database_operations,
    )

    config = {
        # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
        "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
        "echo": False,
        "future": True,
        # "pool_pre_ping": True,
        # "pool_size": 10,
        # "max_overflow": 10,
        "pool_recycle": 3600,
        # "pool_timeout": 30,
    }

    db_config = database_config.DBConfig(config)

    async_db = async_database.AsyncDatabase(db_config)

    db_ops = database_operations.DatabaseOperations(async_db)

    ```
    """
    # 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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)
# count query
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)
        # count query
        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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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 many records
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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 many records
        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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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
        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:

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)
# Delete multiple records
deleted_count = await db_ops.delete_many(User, 'id', [1, 2, 3])
print(f"Deleted {deleted_count} records.")

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
    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)
    # Delete multiple records
    deleted_count = await db_ops.delete_many(User, 'id', [1, 2, 3])
    print(f"Deleted {deleted_count} records.")
    ```
    """
    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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)
# delete one record
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)
        # delete one record
        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 {
                    "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) async

Executes multiple non-read SQL queries asynchronously within a single transaction.

This method executes a list of SQL statements that modify the database, such as multiple INSERTs, UPDATEs, or DELETEs. All queries are executed within the same transaction, which is committed if all succeed, or rolled back if any fail.

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, Dict[str, str]]	Union[str, Dict[str, str]]: "complete" if all queries executed and committed successfully,
Union[str, Dict[str, str]]	or an error dictionary if an exception occurred.

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]]]]
) -> Union[str, Dict[str, str]]:
    """
    Executes multiple non-read SQL queries asynchronously within a single transaction.

    This method executes a list of SQL statements that modify the database,
    such as multiple INSERTs, UPDATEs, or DELETEs. All queries are executed
    within the same transaction, which is committed if all succeed, or rolled
    back if any fail.

    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:
        Union[str, Dict[str, str]]: "complete" if all queries executed and committed successfully,
        or an error dictionary if an exception occurred.

    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:
        async with self.async_db.get_db_session() as session:
            for query, values in queries:
                logger.debug(f"Executing query: {query}")
                await session.execute(query, params=values)
            await session.commit()
            logger.debug("All queries executed successfully")
            return "complete"
    except Exception as ex:
        logger.error(f"Exception occurred: {ex}")
        return handle_exceptions(ex)

execute_one(query, values=None) async

Executes a single non-read SQL query asynchronously.

This method executes a single SQL statement that modifies the database, such as INSERT, UPDATE, or DELETE. It handles the execution within an asynchronous session and commits the transaction upon success.

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

Returns:

Type	Description
Union[str, Dict[str, str]]	Union[str, Dict[str, str]]: "complete" if the query executed and committed successfully,
Union[str, Dict[str, str]]	or an error dictionary if an exception occurred.

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
) -> Union[str, Dict[str, str]]:
    """
    Executes a single non-read SQL query asynchronously.

    This method executes a single SQL statement that modifies the database,
    such as INSERT, UPDATE, or DELETE. It handles the execution within an
    asynchronous session and commits the transaction upon success.

    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.

    Returns:
        Union[str, Dict[str, str]]: "complete" if the query executed and committed successfully,
        or an error dictionary if an exception occurred.

    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}")
            await session.execute(query, params=values)
            await session.commit()
            logger.debug("Query executed successfully")
            return "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:

from sqlalchemy import Table, MetaData, Column,
Integer, String from dsg_lib.async_database_functions import module_name metadata = MetaData()
my_table = Table('my_table', metadata,
                Column('id', Integer, primary_key=True), Column('name',
                String))

from dsg_lib.async_database_functions import (
    async_database,
    base_schema,
    database_config,
    database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)
# get columns details
columns = await db_ops.get_columns_details(my_table)

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
    from sqlalchemy import Table, MetaData, Column,
    Integer, String from dsg_lib.async_database_functions import module_name metadata = MetaData()
    my_table = Table('my_table', metadata,
                    Column('id', Integer, primary_key=True), Column('name',
                    String))

    from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
    )
    # Create a DBConfig instance
    config = {
        # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
        "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
        "echo": False,
        "future": True,
        # "pool_pre_ping": True,
        # "pool_size": 10,
        # "max_overflow": 10,
        "pool_recycle": 3600,
        # "pool_timeout": 30,
    }
    # 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)
    # get columns details
    columns = await db_ops.get_columns_details(my_table)
    ```
    """
    # 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

from sqlalchemy import Table, MetaData, Column, Integer,
    String from dsg_lib.async_database_functions import module_name metadata = MetaData()
    my_table = Table('my_table', metadata,
                    Column('id', Integer, primary_key=True),
                    Column('name', String, primary_key=True))
from dsg_lib.async_database_functions import (
    async_database,
    base_schema,
    database_config,
    database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)

# get primary keys
primary_keys = await db_ops.get_primary_keys(my_table)

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
        from sqlalchemy import Table, MetaData, Column, Integer,
            String from dsg_lib.async_database_functions import module_name metadata = MetaData()
            my_table = Table('my_table', metadata,
                            Column('id', Integer, primary_key=True),
                            Column('name', String, primary_key=True))
        from dsg_lib.async_database_functions import (
            async_database,
            base_schema,
            database_config,
            database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)

        # get primary keys
        primary_keys = await db_ops.get_primary_keys(my_table)
        ```
    """
    # 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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)
# get table names
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)
        # get table names
        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

read_multi_query(queries) async

Executes multiple fetch queries on the database and returns a dictionary of results for each query.

This asynchronous method takes a dictionary where each key is a query name and each value is a SQLAlchemy Select query object. The method executes each query and returns a dictionary where each key is the query name, and the corresponding value is a list of records that match that query.

Parameters:

Name	Type	Description	Default
queries	Dict[str, Select]	A dictionary of SQLAlchemy Select	required

Returns:

Name	Type	Description
dict		A dictionary where each key is a query name and each value is
		a list of records that match the query.

Raises:

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

Example

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)
# read multi query
queries = {
    "query1": select(User).where(User.age > 30),
    "query2": select(User).where(User.age < 20),
}
results = await db_ops.read_multi_query(queries)

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 on the database and returns a dictionary
    of results for each query.

    This asynchronous method takes a dictionary where each key is a query
    name and each value is a SQLAlchemy `Select` query object. The method executes each
    query and returns a dictionary where each key is the query name, and the
    corresponding value is a list of records that match that query.

    Parameters:
        queries (Dict[str, Select]): A dictionary of SQLAlchemy `Select`
        query objects.

    Returns:
        dict: A dictionary where each key is a query name and each value is
        a list of records that match the query.

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

    Example:
        ```python
        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)
        # read multi query
        queries = {
            "query1": select(User).where(User.age > 30),
            "query2": select(User).where(User.age < 20),
        }
        results = await db_ops.read_multi_query(queries)
        ```
    """
    # Log the start of the operation
    logger.debug("Starting read_multi_query operation")

    try:
        results = {}
        # Start a new database session
        async with self.async_db.get_db_session() as session:
            for query_name, query in queries.items():
                # 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)
                data = result.scalars().all()

                # Convert the records to dictionaries for logging
                data_dicts = [record.__dict__ for record in data]
                logger.debug(f"Fetch result for query '{query_name}': {data_dicts}")

                # Store the records in the results dictionary
                results[query_name] = data
        return results

    except Exception as ex:
        # Handle any exceptions that occur during the query execution
        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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)
# read one record
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)
        # read one record
        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

read_query(query) async

Executes a fetch query on the database and returns a list of records that match the query.

This asynchronous method accepts a SQLAlchemy Select query object. It returns a list of records that match the query.

Parameters:

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

Returns:

Name	Type	Description
list		A list of records that match the query.

Raises:

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

Example

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)
# read query
records = await db_ops.read_query(select(User).where(User.age > 30))

Source code in dsg_lib/async_database_functions/database_operations.py

async def read_query(self, query):
    """
    Executes a fetch query on the database and returns a list of records
    that match the query.

    This asynchronous method accepts a SQLAlchemy `Select` query object.
    It returns a list of records that match the query.

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

    Returns:
        list: A list of records that match the query.

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

    Example:
        ```python
        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)
        # read query
        records = await db_ops.read_query(select(User).where(User.age > 30))
        ```
    """
    # 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 = result.scalars().all()
            logger.debug(f"read_query result: {records}")
            # Log the successful query execution
            if all(
                isinstance(record, tuple) for record in records
            ):  # pragma: no cover
                logger.debug(f"read_query result is a tuple {type(records)}")
                # If all records are tuples, convert them to dictionaries
                records_data = [
                    dict(zip(("request_group_id", "count"), record, strict=False))
                    for record in records
                ]
            else:
                logger.debug(f"read_query result is a dictionary {type(records)}")
                # Otherwise, try to convert the records to dictionaries using the __dict__ attribute
                records_data = [record.__dict__ for record in records]

            logger.debug(
                f"Fetch query executed successfully. Records: {records_data}"
            )

            return records

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

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

from dsg_lib.async_database_functions import (
async_database,
base_schema,
database_config,
database_operations,
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}
# 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)
# update one record
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
        from dsg_lib.async_database_functions import (
        async_database,
        base_schema,
        database_config,
        database_operations,
        )
        # Create a DBConfig instance
        config = {
            # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
            "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
            "echo": False,
            "future": True,
            # "pool_pre_ping": True,
            # "pool_size": 10,
            # "max_overflow": 10,
            "pool_recycle": 3600,
            # "pool_timeout": 30,
        }
        # 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)
        # update one record
        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 {
                    "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 dictionary containing the error details.

Parameters:

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

Returns:

Name	Type	Description
dict	Dict[str, str]	A dictionary containing the error details. The dictionary has two
keys	Dict[str, str]	'error' and 'details'.

Example:

from dsg_lib.async_database_functions import database_operations

try:
    # Some database operation that might raise an exception pass
except Exception as ex:
    error_details = database_operations.handle_exceptions(ex)
    print(error_details)

Source code in dsg_lib/async_database_functions/database_operations.py

def handle_exceptions(ex: Exception) -> Dict[str, str]:
    """
    Handles exceptions for database operations.

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

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

    Returns:
        dict: A dictionary containing the error details. The dictionary has two
        keys: 'error' and 'details'.

    Example:
    ```python
    from dsg_lib.async_database_functions import database_operations

    try:
        # Some database operation that might raise an exception pass
    except Exception as ex:
        error_details = database_operations.handle_exceptions(ex)
        print(error_details)
    ```
    """
    # 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 {"error": "IntegrityError", "details": error_only}
    elif isinstance(ex, SQLAlchemyError):
        # Log the error and return the error details
        logger.error(f"SQLAlchemyError occurred: {ex}")
        return {"error": "SQLAlchemyError", "details": error_only}
    else:
        # Log the error and return the error details
        logger.error(f"Exception occurred: {ex}")
        return {"error": "General Exception", "details": str(ex)}

Ended: Database Functions

Recipes

Full Example of FastAPI with Aync Database and Endpoints

You can find this in the examples folder of the repository.

Install dependencies

pip install dsg_lib[all] tqdm

Make App

Copy the fastapi code below after installing. (assumption is main.py)

# -*- coding: utf-8 -*-
"""
Author: Mike Ryan
Date: 2024/05/16
License: MIT
"""
import datetime
import secrets
import time
from contextlib import asynccontextmanager

from fastapi import Body, FastAPI, Query
from fastapi.responses import RedirectResponse
# from loguru import logger
# import logging as logger
from . import logger
from pydantic import BaseModel, EmailStr
from sqlalchemy import Column, ForeignKey, Select, String
from sqlalchemy.orm import relationship
from tqdm import tqdm

from dsg_lib.async_database_functions import (
    async_database,
    base_schema,
    database_config,
    database_operations,
)
from dsg_lib.common_functions import logging_config
from dsg_lib.fastapi_functions import system_health_endpoints  # , system_tools_endpoints

logging_config.config_log(
    logging_level="INFO", log_serializer=False, log_name="log.log"
)
# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}

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


class User(base_schema.SchemaBaseSQLite, async_db.Base):
    """
    User table storing user details like first name, last name, and email
    """

    __tablename__ = "users"
    __table_args__ = {
        "comment": "User table storing user details like first name, last name, and email"
    }

    first_name = Column(String(50), unique=False, index=True)  # First name of the user
    last_name = Column(String(50), unique=False, index=True)  # Last name of the user
    email = Column(
        String(200), unique=True, index=True, nullable=True
    )  # Email of the user, must be unique
    addresses = relationship(
        "Address", order_by="Address.pkid", back_populates="user"
    )  # Relationship to the Address class


class Address(base_schema.SchemaBaseSQLite, async_db.Base):
    """
    Address table storing address details like street, city, and zip code
    """

    __tablename__ = "addresses"
    __table_args__ = {
        "comment": "Address table storing address details like street, city, and zip code"
    }

    street = Column(String(200), unique=False, index=True)  # Street of the address
    city = Column(String(200), unique=False, index=True)  # City of the address
    zip = Column(String(50), unique=False, index=True)  # Zip code of the address
    user_id = Column(
        String(36), ForeignKey("users.pkid")
    )  # Foreign key to the User table
    user = relationship(
        "User", back_populates="addresses"
    )  # Relationship to the User class

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("starting up")
    # Create the tables in the database
    await async_db.create_tables()

    create_users = True
    if create_users:
        await create_a_bunch_of_users(single_entry=2000, many_entries=20000)
    yield
    logger.info("shutting down")
    await async_db.disconnect()
    logger.info("database disconnected")
    print("That's all folks!")



# Create an instance of the FastAPI class
app = FastAPI(
    title="FastAPI Example",  # The title of the API
    description="This is an example of a FastAPI application using the DevSetGo Toolkit.",  # A brief description of the API
    version="0.1.0",  # The version of the API
    docs_url="/docs",  # The URL where the API documentation will be served
    redoc_url="/redoc",  # The URL where the ReDoc documentation will be served
    openapi_url="/openapi.json",  # The URL where the OpenAPI schema will be served
    debug=True,  # Enable debug mode
    middleware=[],  # A list of middleware to include in the application
    routes=[],  # A list of routes to include in the application
    lifespan=lifespan,  # this is the replacement for the startup and shutdown events
)


@app.get("/")
async def root():
    """
    Root endpoint of API
    Returns:
        Redrects to openapi document
    """
    # redirect to openapi docs
    logger.info("Redirecting to OpenAPI docs")
    response = RedirectResponse(url="/docs")
    return response


config_health = {
    "enable_status_endpoint": True,
    "enable_uptime_endpoint": True,
    "enable_heapdump_endpoint": True,
}
app.include_router(
    system_health_endpoints.create_health_router(config=config_health),
    prefix="/api/health",
    tags=["system-health"],
)




async def create_a_bunch_of_users(single_entry=0, many_entries=0):
    logger.info(f"single_entry: {single_entry}")
    await async_db.create_tables()
    # Create a list to hold the user data

    # Create a loop to generate user data

    for _ in tqdm(range(single_entry), desc="executing one"):
        value = secrets.token_hex(16)
        user = User(
            first_name=f"First{value}",
            last_name=f"Last{value}",
            email=f"user{value}@example.com",
        )
        logger.info(f"created_users: {user}")
        await db_ops.create_one(user)

    users = []
    # Create a loop to generate user data
    for i in tqdm(range(many_entries), desc="executing many"):
        value_one = secrets.token_hex(4)
        value_two = secrets.token_hex(8)
        user = User(
            first_name=f"First{value_one}{i}{value_two}",
            last_name=f"Last{value_one}{i}{value_two}",
            email=f"user{value_one}{i}{value_two}@example.com",
        )
        logger.info(f"created_users: {user.first_name}")
        users.append(user)

    # Use db_ops to add the users to the database
    await db_ops.create_many(users)


@app.get("/database/get-primary-key", tags=["Database Examples"])
async def table_primary_key():
    logger.info("Getting primary key of User table")
    pk = await db_ops.get_primary_keys(User)
    logger.info(f"Primary key of User table: {pk}")
    return {"pk": pk}


@app.get("/database/get-column-details", tags=["Database Examples"])
async def table_column_details():
    logger.info("Getting column details of User table")
    columns = await db_ops.get_columns_details(User)
    logger.info(f"Column details of User table: {columns}")
    return {"columns": columns}


@app.get("/database/get-tables", tags=["Database Examples"])
async def table_table_details():
    logger.info("Getting table names")
    tables = await db_ops.get_table_names()
    logger.info(f"Table names: {tables}")
    return {"table_names": tables}


@app.get("/database/get-count", tags=["Database Examples"])
async def get_count():
    logger.info("Getting count of users")
    count = await db_ops.count_query(Select(User))
    logger.info(f"Count of users: {count}")
    return {"count": count}


@app.get("/database/get-all", tags=["Database Examples"])
async def get_all(offset: int = 0, limit: int = Query(100, le=100000, ge=1)):
    logger.info(f"Getting all users with offset {offset} and limit {limit}")
    records = await db_ops.read_query(Select(User).offset(offset).limit(limit))
    logger.info(f"Retrieved {len(records)} users")
    return {"records": records}


@app.get("/database/get-one-record", tags=["Database Examples"])
async def read_one_record(record_id: str):
    logger.info(f"Reading one record with id {record_id}")
    record = await db_ops.read_one_record(Select(User).where(User.pkid == record_id))
    logger.info(f"Record with id {record_id}: {record}")
    return record


class UserBase(BaseModel):
    first_name: str
    last_name: str
    email: EmailStr


class UserCreate(UserBase):
    pass


@app.post("/database/create-one-record", status_code=201, tags=["Database Examples"])
async def create_one_record(new_user: UserCreate):
    logger.info(f"Creating one record: {new_user}")
    user = User(**new_user.dict())
    record = await db_ops.create_one(user)
    logger.info(f"Created record: {record}")
    return record


@app.post("/database/create-many-records", status_code=201, tags=["Database Examples"])
async def create_many_records(number_of_users: int = Query(100, le=1000, ge=1)):
    logger.info(f"Creating {number_of_users} records")
    t0 = time.time()
    users = []
    # Create a loop to generate user data
    for i in tqdm(range(number_of_users), desc="executing many"):
        value_one = secrets.token_hex(4)
        value_two = secrets.token_hex(8)
        user = User(
            first_name=f"First{value_one}{i}{value_two}",
            last_name=f"Last{value_one}{i}{value_two}",
            email=f"user{value_one}{i}{value_two}@example.com",
        )
        logger.info(f"Created user: {user.first_name}")
        users.append(user)

    # Use db_ops to add the users to the database
    await db_ops.create_many(users)
    t1 = time.time()
    process_time = format(t1 - t0, ".4f")
    logger.info(f"Created {number_of_users} records in {process_time} seconds")
    return {"number_of_users": number_of_users, "process_time": process_time}


@app.put("/database/update-one-record", status_code=200, tags=["Database Examples"])
async def update_one_record(
    id: str = Body(
        ...,
        description="UUID to update",
        examples=["6087cce8-0bdd-48c2-ba96-7d557dae843e"],
    ),
    first_name: str = Body(..., examples=["Agent"]),
    last_name: str = Body(..., examples=["Smith"]),
    email: str = Body(..., examples=["jim@something.com"]),
):
    logger.info(f"Updating one record with id {id}")
    # adding date_updated to new_values as it is not supported in sqlite \
    # and other database may not either.
    new_values = {
        "first_name": first_name,
        "last_name": last_name,
        "email": email,
        "date_updated": datetime.datetime.now(datetime.timezone.utc),
    }
    record = await db_ops.update_one(table=User, record_id=id, new_values=new_values)
    logger.info(f"Updated record with id {id}")
    return record


@app.delete("/database/delete-one-record", status_code=200, tags=["Database Examples"])
async def delete_one_record(record_id: str = Body(...)):
    logger.info(f"Deleting one record with id {record_id}")
    record = await db_ops.delete_one(table=User, record_id=record_id)
    logger.info(f"Deleted record with id {record_id}")
    return record


@app.delete(
    "/database/delete-many-records-aka-this-is-a-bad-idea",
    status_code=201,
    tags=["Database Examples"],
)
async def delete_many_records(
    id_values: list = Body(...), id_column_name: str = "pkid"
):
    logger.info(f"Deleting many records with ids {id_values}")
    record = await db_ops.delete_many(
        table=User, id_column_name="pkid", id_values=id_values
    )
    logger.info(f"Deleted records with ids {id_values}")
    return record


@app.get(
    "/database/get-list-of-records-to-paste-into-delete-many-records",
    tags=["Database Examples"],
)
async def read_list_of_records(
    offset: int = Query(0, le=1000, ge=0), limit: int = Query(100, le=10000, ge=1)
):
    logger.info(f"Reading list of records with offset {offset} and limit {limit}")
    records = await db_ops.read_query(Select(User), offset=offset, limit=limit)
    records_list = []
    for record in records:
        records_list.append(record.pkid)
    logger.info(f"Read list of records: {records_list}")
    return records_list


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=5000)

Run Code

In the console (linux) run the code below. Open browser to http://127.0.0.1:5000 to see app.

python3 main.py

Examples

Here are a few examples of how to use the database functions

Asyncio Script Example

Example of how to use in a script

import asyncio
from sqlalchemy import select
from dsg_lib.async_database_functions import database_config, async_database, database_operations

# Configuration
config = {
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}

# Create a DBConfig instance
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)

# User class
class User(async_db.Base):
    __tablename__ = "users"
    first_name = Column(String, unique=False, index=True)
    last_name = Column(String, unique=False, index=True)
    email = Column(String, unique=True, index=True, nullable=True)

# Async function to get all users
async def get_all_users():
    # Create a select query
    query = select(User)

    # Execute the query and fetch all results
    users = await db_ops.read_query(query)

    # Print the users
    for user in users:
        print(f"User: {user.first_name} {user.last_name}, Email: {user.email}")

# Run the async function
asyncio.run(get_all_users())

FastAPI Example

# -*- coding: utf-8 -*-

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from loguru import logger
from tqdm import tqdm

from dsg_lib import logging_config

logging_config.config_log(
    logging_level="Debug", log_serializer=False, log_name="log.log"
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("starting up")
    # Create the tables in the database
    await async_db.create_tables()

    create_users = True
    if create_users:
        await create_a_bunch_of_users(single_entry=23, many_entries=100)
    yield
    logger.info("shutting down")


# Create an instance of the FastAPI class
app = FastAPI(
    title="FastAPI Example",  # The title of the API
    description="This is an example of a FastAPI application using the DevSetGo Toolkit.",  # A brief description of the API
    version="0.1.0",  # The version of the API
    docs_url="/docs",  # The URL where the API documentation will be served
    redoc_url="/redoc",  # The URL where the ReDoc documentation will be served
    openapi_url="/openapi.json",  # The URL where the OpenAPI schema will be served
    debug=True,  # Enable debug mode
    middleware=[],  # A list of middleware to include in the application
    routes=[],  # A list of routes to include in the application
    lifespan=lifespan,
)


@app.get("/")
async def root():
    """
    Root endpoint of API
    Returns:
        Redrects to openapi document
    """
    # redirect to openapi docs
    logger.info("Redirecting to OpenAPI docs")
    response = RedirectResponse(url="/docs")
    return response

from sqlalchemy import Column, Delete, Select, String, Update

from dsg_lib import (
    async_database,
    base_schema,
    database_config,
    database_operations,
)

# Create a DBConfig instance
config = {
    # "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    # "pool_pre_ping": True,
    # "pool_size": 10,
    # "max_overflow": 10,
    "pool_recycle": 3600,
    # "pool_timeout": 30,
}

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)


# User class inherits from SchemaBase and async_db.Base
# This class represents the User table in the database
class User(base_schema.SchemaBase, async_db.Base):
    __tablename__ = "users"  # Name of the table in the database

    # Define the columns of the table
    first_name = Column(String, unique=False, index=True)  # First name of the user
    last_name = Column(String, unique=False, index=True)  # Last name of the user
    email = Column(
        String, unique=True, index=True, nullable=True
    )  # Email of the user, must be unique

async def create_a_bunch_of_users(single_entry=0, many_entries=0):
    logger.info(f"single_entry: {single_entry}")
    await async_db.create_tables()
    # Create a list to hold the user data

    # Create a loop to generate user data

    for i in tqdm(range(single_entry), desc="executing one"):
        value = secrets.token_hex(16)
        user = User(
            first_name=f"First{value}",
            last_name=f"Last{value}",
            email=f"user{value}@example.com",
        )
        logger.info(f"created_users: {user}")
        await db_ops.create_one(user)

    users = []
    # Create a loop to generate user data
    for i in tqdm(range(many_entries), desc="executing many"):
        value_one = secrets.token_hex(4)
        value_two = secrets.token_hex(8)
        user = User(
            first_name=f"First{value_one}{i}{value_two}",
            last_name=f"Last{value_one}{i}{value_two}",
            email=f"user{value_one}{i}{value_two}@example.com",
        )
        logger.info(f"created_users: {user.first_name}")
        users.append(user)

    # Use db_ops to add the users to the database
    await db_ops.create_many(users)


@app.get("/database/get-count")
async def get_count():
    count = await db_ops.count_query(Select(User))
    return {"count": count}


# endpoint to get list of user
@app.get("/database/get-all")
async def get_all(offset: int = 0, limit: int = 100):
    records = await db_ops.read_query(Select(User).offset(offset).limit(limit))
    return {"records": records}


@app.get("/database/get-primary-key")
async def table_primary_key():
    pk = await db_ops.get_primary_keys(User)
    return {"pk": pk}


@app.get("/database/get-column-details")
async def table_column_details():
    columns = await db_ops.get_columns_details(User)
    return {"columns": columns}


@app.get("/database/get-tables")
async def table_table_details():
    tables = await db_ops.get_table_names()
    return {"table_names": tables}


@app.get("/database/get-one-record")
async def get_one_record(record_id: str):
    record = await db_ops.get_one_record(Select(User).where(User.pkid == record_id))
    return {"record": record}

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=5000)

Configuration Examples

# SQLite in memory database
config = {
    "database_uri": "sqlite+aiosqlite:///:memory:?cache=shared",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}

# PostgreSQL database
config = {
    "database_uri": "postgresql+asyncpg://postgres:postgres@postgresdb/postgres",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}

# MySQL database
config = {
    "database_uri": "mysql+aiomysql://root:root@localhost/test",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}

# SQL Server database
config = {
    "database_uri": "mssql+aiomssql://sa:yourStrong(!)Password@localhost:1433/master",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}

# Oracle database
config = {
    "database_uri": "oracle+oracledb_async://scott:tiger@localhost/?service_name=XEPDB1",
    "echo": False,
    "future": True,
    "pool_recycle": 3600,
}

Logging Example

# -*- coding: utf-8 -*-
"""
Author: Mike Ryan
Date: 2024/05/16
License: MIT
"""
import logging
import multiprocessing
import secrets
import threading

# from loguru import logger
# import logging as logger
from . import logger
from tqdm import tqdm

from dsg_lib.common_functions import logging_config

# Configure logging as before
logging_config.config_log(
    logging_directory='log',
    log_name='log',
    logging_level='DEBUG',
    log_rotation='100 MB',
    log_retention='10 days',
    log_backtrace=True,
    log_serializer=True,
    log_diagnose=True,
    # app_name='my_app',
    # append_app_name=True,
    intercept_standard_logging=True,
    enqueue=True,
)


# @logger.catch
def div_zero(x, y):
    try:
        return x / y
    except ZeroDivisionError as e:
        logger.error(f'{e}')
        logging.error(f'{e}')


# @logger.catch
def div_zero_two(x, y):
    try:
        return x / y
    except ZeroDivisionError as e:
        logger.error(f'{e}')
        logging.error(f'{e}')


def log_big_string(lqty=100, size=256):
    big_string = secrets.token_urlsafe(size)
    for _ in range(lqty):
        logging.debug(f'Lets make this a big message {big_string}')
        div_zero(x=1, y=0)
        div_zero_two(x=1, y=0)
        # after configuring logging
        # use loguru to log messages
        logger.debug('This is a loguru debug message')
        logger.info('This is an loguru info message')
        logger.error('This is an loguru error message')
        logger.warning('This is a loguru warning message')
        logger.critical('This is a loguru critical message')

        # will intercept all standard logging messages also
        logging.debug('This is a standard logging debug message')
        logging.info('This is an standard logging info message')
        logging.error('This is an standard logging error message')
        logging.warning('This is a standard logging warning message')
        logging.critical('This is a standard logging critical message')


def worker(wqty=1000, lqty=100, size=256):
    for _ in tqdm(range(wqty), ascii=True, leave=True):  # Adjusted for demonstration
        log_big_string(lqty=lqty, size=size)


def main(wqty: int = 100, lqty: int = 10, size: int = 256, workers: int = 16, thread_test: bool = False, process_test: bool = False):
    if process_test:
        processes = []
        # Create worker processes
        for _ in tqdm(range(workers), desc="Multi-Processing Start", leave=True):
            p = multiprocessing.Process(
                target=worker, args=(wqty, lqty, size,))
            processes.append(p)
            p.start()

        for p in tqdm((processes), desc="Multi-Processing Start", leave=False):
            p.join(timeout=60)  # Timeout after 60 seconds
            if p.is_alive():
                logger.error(f"Process {p.name} is hanging. Terminating.")
                p.terminate()
                p.join()

    if thread_test:
        threads = []
        for _ in tqdm(range(workers), desc="Threading Start", leave=True):  # Create worker threads
            t = threading.Thread(target=worker, args=(wqty, lqty, size,))
            threads.append(t)
            t.start()

        for t in tqdm(threads, desc="Threading Gather", leave=False):
            t.join()


if __name__ == "__main__":
    from time import time
    start = time()
    main(wqty=5, lqty=50, size=64, workers=8, thread_test=False, process_test=True)
    print(f"Execution time: {time()-start:.2f} seconds")

Patterns

# -*- coding: utf-8 -*-
"""
Author: Mike Ryan
Date: 2024/05/16
License: MIT
"""
import pprint
from random import randint

from dsg_lib.common_functions.patterns import pattern_between_two_char

ASCII_LIST = [
    " ",
    "!",
    '""',
    "#",
    "$",
    "%",
    "&",
    "'",
    "(",
    ")",
    "*",
    "+",
    ",",
    "-",
    ".",
    "/",
    "0",
    "1",
    "2",
    "3",
    "4",
    "5",
    "6",
    "7",
    "8",
    "9",
    ":",
    ";",
    "<",
    "=",
    ">",
    "?",
    "@",
    "A",
    "B",
    "C",
    "D",
    "E",
    "F",
    "G",
    "H",
    "I",
    "J",
    "K",
    "L",
    "M",
    "N",
    "O",
    "P",
    "Q",
    "R",
    "S",
    "T",
    "U",
    "V",
    "W",
    "X",
    "Y",
    "Z",
    "[",
    "\\",
    "]",
    "^",
    "_",
    "`",
    "a",
    "b",
    "c",
    "d",
    "e",
    "f",
    "g",
    "h",
    "i",
    "j",
    "k",
    "l",
    "m",
    "n",
    "o",
    "p",
    "q",
    "r",
    "s",
    "t",
    "u",
    "v",
    "w",
    "x",
    "y",
    "z",
    "{",
    "|",
    "}",
    "~",
    "€",
    "‚",
    "ƒ",
    "„",
    "…",
    "†",
    "‡",
    "ˆ",
    "‰",
    "Š",
    "‹",
    "Œ",
    "Ž",
    "‘",
    "’",
    "“",
    "”",
    "•",
    "–",
    "—",
    "˜",
    "™",
    "š",
    "›",
    "œ",
    "ž",
    "Ÿ",
    "¡",
    "¢",
    "£",
    "¤",
    "¥",
    "¦",
    "§",
    "¨",
    "©",
    "ª",
    "«",
    "¬",
    "®",
    "¯",
    "°",
    "±",
    "²",
    "³",
    "´",
    "µ",
    "¶",
    "·",
    "¸",
    "¹",
    "º",
    "»",
    "¼",
    "½",
    "¾",
    "¿",
    "À",
    "Á",
    "Â",
    "Ã",
    "Ä",
    "Å",
    "Æ",
    "Ç",
    "È",
    "É",
    "Ê",
    "Ë",
    "Ì",
    "Í",
    "Î",
    "Ï",
    "Ð",
    "Ñ",
    "Ò",
    "Ó",
    "Ô",
    "Õ",
    "Ö",
    "×",
    "Ø",
    "Ù",
    "Ú",
    "Û",
    "Ü",
    "Ý",
    "Þ",
    "ß",
    "à",
    "á",
    "â",
    "ã",
    "ä",
    "å",
    "æ",
    "ç",
    "è",
    "é",
    "ê",
    "ë",
    "ì",
    "í",
    "î",
    "ï",
    "ð",
    "ñ",
    "ò",
    "ó",
    "ô",
    "õ",
    "ö",
    "÷",
    "ø",
    "ù",
    "ú",
    "û",
    "ü",
    "ý",
    "þ",
    "ÿ",
]

pp = pprint.PrettyPrinter(indent=4)


def pattern_find(left_char: str, right_char: str, text_block: str):
    data = pattern_between_two_char(text_block, left_char, right_char)
    pp.pprint(data)


def run_examples():
    text_block = "Lfound oneR Lfound twoR"
    left_char = "L"
    right_char = "R"
    pattern_find(left_char=left_char, right_char=right_char, text_block=text_block)

    for _ in range(100):
        long_input = "xyz" * randint(100, 100000)
        long_text = f"{long_input}abc<one>123<two>456<three>{long_input}"

        result = pattern_between_two_char(
            text_string=long_text, left_characters="<", right_characters=">"
        )
        print(result["found"])


if __name__ == "__main__":
    run_examples()

Validating Email Addresses

Example of how to use in a script

# -*- coding: utf-8 -*-
"""
This module is used to validate a list of email addresses using various configurations.

The module imports the `validate_email_address` function from the `dsg_lib.common_functions.email_validation`
module and uses it to validate a list of email addresses. The email addresses and configurations are hard-coded
into the module.

The module measures the time taken to validate all the email addresses with all the configurations and prints
the results in a sorted order.

The module can be run as a standalone script.

Example:
    $ python validate_emails.py

Attributes:
    email_addresses (list of str): A list of email addresses to validate.
    configurations (list of dict): A list of configurations to use for validation. Each configuration is a
    dictionary with the following keys:
        - check_deliverability (bool): Whether to check if the email address is deliverable.
        - test_environment (bool): Whether the function is being run in a test environment.
        - allow_smtputf8 (bool): Whether to allow non-ASCII characters in the email address.
        - allow_empty_local (bool): Whether to allow email addresses with an empty local part.
        - allow_quoted_local (bool): Whether to allow email addresses with a quoted local part.
        - allow_display_name (bool): Whether to allow email addresses with a display name.
        - allow_domain_literal (bool): Whether to allow email addresses with a domain literal.
        - globally_deliverable (bool): Whether the email address should be globally deliverable.
        - timeout (int): The timeout for the validation in seconds.
        - dns_type (str): The type of DNS to use for the validation. Can be 'dns' or 'timeout'.

Functions:
    validate_email_address(email: str, **kwargs: dict) -> dict: Validates an email address using the provided
    configuration and returns a dictionary with the results.

Author: Mike Ryan
Date: 2024/05/16
License: MIT
"""
import pprint
import time

from dsg_lib.common_functions.email_validation import validate_email_address

if __name__ == "__main__":

    # create a list of email addresses to check if valid
    email_addresses = [
        "bob@devsetgo.com",
        "bob@devset.go",
        "foo@yahoo.com",
        "bob@gmail.com",
        "very fake@devsetgo.com",
        "jane.doe@example.com",
        "john_doe@example.co.uk",
        "user.name+tag+sorting@example.com",
        "x@example.com",  # shortest possible email address
        "example-indeed@strange-example.com",
        "admin@mailserver1",  # local domain name with no TLD
        "example@s.example",  # see the list of Internet top-level domains
        '" "@example.org',  # space between the quotes
        '"john..doe"@example.org',  # quoted double dot
        "mailhost!username@example.org",  # bangified host route used for uucp mailers
        "user%example.com@example.org",  # percent sign in local part
        "user-@example.org",  # valid due to the last character being an allowed character
        # Invalid email addresses
        "Abc.example.com",  # no @ character
        "A@b@c@example.com",  # only one @ is allowed outside quotation marks
        'a"b(c)d,e:f;g<h>i[j\\k]l@example.com',  # none of the special characters in this local part are allowed outside quotation marks
        'just"not"right@example.com',  # quoted strings must be dot separated or the only element making up the local-part
        'this is"not\\allowed@example.com',  # spaces, quotes, and backslashes may only exist when within quoted strings and preceded by a backslash
        'this\\ still\\"not\\\\allowed@example.com',  # even if escaped (preceded by a backslash), spaces, quotes, and backslashes must still be contained by quotes
        "1234567890123456789012345678901234567890123456789012345678901234+x@example.com",  # local part is longer than 64 characters
        # Emails with empty local part
        "@example.com",  # only valid if allow_empty_local is True
        # Emails with non-ASCII characters
        "üñîçøðé@example.com",  # only valid if allow_smtputf8 is True
        "user@üñîçøðé.com",  # only valid if allow_smtputf8 is True
        # Emails with quoted local part
        '"john.doe"@example.com',  # only valid if allow_quoted_local is True
        '"john..doe"@example.com',  # only valid if allow_quoted_local is True
        # Emails with display name
        "John Doe <john@example.com>",  # only valid if allow_display_name is True
        # Emails with domain literal
        "user@[192.0.2.1]",  # only valid if allow_domain_literal is True
        # Emails with long local part
        "a" * 65 + "@example.com",  # local part is longer than 64 characters
        # Emails with invalid characters
        "john doe@example.com",  # space is not allowed
        "john@doe@example.com",  # only one @ is allowed
        "john.doe@.com",  # domain can't start with a dot
        "john.doe@example..com",  # domain can't have two consecutive dots
        "test@google.com",
    ]

    # create a list of configurations
    configurations = [
        {
            "check_deliverability": True,
            "test_environment": False,
            "allow_smtputf8": False,
            "allow_empty_local": False,
            "allow_quoted_local": False,
            "allow_display_name": False,
            "allow_domain_literal": False,
            "globally_deliverable": None,
            "timeout": 10,
            "dns_type": "timeout",
        },
        {
            "check_deliverability": False,
            "test_environment": True,
            "allow_smtputf8": True,
            "allow_empty_local": True,
            "allow_quoted_local": True,
            "allow_display_name": True,
            "allow_domain_literal": True,
            "globally_deliverable": None,
            "timeout": 5,
            "dns_type": "dns",
        },
        {"check_deliverability": True},
        {
            "check_deliverability": False,
            "test_environment": False,
            "allow_smtputf8": True,
            "allow_empty_local": False,
            "allow_quoted_local": True,
            "allow_display_name": False,
            "allow_domain_literal": True,
            "globally_deliverable": None,
            "timeout": 15,
            "dns_type": "timeout",
        },
        {
            "check_deliverability": True,
            "test_environment": True,
            "allow_smtputf8": False,
            "allow_empty_local": True,
            "allow_quoted_local": False,
            "allow_display_name": True,
            "allow_domain_literal": False,
            "globally_deliverable": None,
            "timeout": 20,
            "dns_type": "dns",
        },
        {
            "check_deliverability": False,
            "test_environment": False,
            "allow_smtputf8": True,
            "allow_empty_local": True,
            "allow_quoted_local": True,
            "allow_display_name": True,
            "allow_domain_literal": True,
            "globally_deliverable": None,
            "timeout": 25,
            "dns_type": "timeout",
        },
        {
            "check_deliverability": True,
            "test_environment": True,
            "allow_smtputf8": False,
            "allow_empty_local": False,
            "allow_quoted_local": False,
            "allow_display_name": False,
            "allow_domain_literal": False,
            "globally_deliverable": None,
            "timeout": 30,
            "dns_type": "dns",
        },
        {
            "check_deliverability": False,
            "test_environment": True,
            "allow_smtputf8": True,
            "allow_empty_local": False,
            "allow_quoted_local": True,
            "allow_display_name": True,
            "allow_domain_literal": False,
            "globally_deliverable": None,
            "timeout": 35,
            "dns_type": "timeout",
        },
        {
            "check_deliverability": True,
            "test_environment": False,
            "allow_smtputf8": False,
            "allow_empty_local": True,
            "allow_quoted_local": True,
            "allow_display_name": False,
            "allow_domain_literal": True,
            "globally_deliverable": None,
            "timeout": 40,
            "dns_type": "dns",
        },
        {
            "check_deliverability": False,
            "test_environment": True,
            "allow_smtputf8": True,
            "allow_empty_local": False,
            "allow_quoted_local": False,
            "allow_display_name": True,
            "allow_domain_literal": True,
            "globally_deliverable": None,
            "timeout": 45,
            "dns_type": "timeout",
        },
    ]

    t0 = time.time()
    validity = []

    for email in email_addresses:
        for config in configurations:

            res = validate_email_address(email, **config)
            validity.append(res)
    t1 = time.time()
    validity = sorted(validity, key=lambda x: x["email"])

    for v in validity:
        pprint.pprint(v, indent=4)

    print(f"Time taken: {t1 - t0:.2f}")

Ended: Recipes

About

About

DevSetGo Library is a collection of tools and utilities to help developers with their projects. It has been primarily developed to help with the development of the DevSetGo.com website and other projects in my personal and professional life. It is a work in progress and will be updated as needed.

The driving force behind the library is to limit the amount of boilerplate code that I (Mike Ryan) have to write for each project. Copying code from one project to another causes issues to get fixed in one project, but not always get updated in others. This library is an attempt to fix that issue and make it easier to maintain code across multiple projects.

The library is written in Python and is available on PyPi. It is open source and available on GitHub. Feel free to use it in your projects and contribute to the library.

About Me

I am a software engineering manager with an eclectic background in various industries (finance, manufacturing, and metrology). I am passionate about software development and love to learn new things.

• DevSetGo.com
• Github

Contributing

Please feel to contribute to this project. Adding common functions is the intent and if you have one to add or improve an existing it is greatly appreciated.

Ways to Contribute!

• Add or improve a function
• Add or improve documentation
• Add or improve Tests
• Report or fix a bug

Ended: About

Changelog

All notable changes to this project will be documented in this file.

The format is based on Keep a Changelog

Latest Changes

Adding new db functions (v2024.11.28.1)

What's Changed

• Adding new general execute queries and adding deprecation (#459) @devsetgo
• pip(deps): bump tox from 4.23.0 to 4.23.2 (#455) @dependabot
• pip(deps): bump fastapi[all] from 0.115.2 to 0.115.4 (#454) @dependabot
• pip(deps): bump tqdm from 4.66.5 to 4.66.6 (#456) @dependabot
• pip(deps): bump pymdown-extensions from 10.11.2 to 10.12 (#457) @dependabot
• pip(deps): bump ruff from 0.7.0 to 0.7.1 (#458) @dependabot

Published Date: 2024 November 28, 22:01

Moving to Calendar Versioning (2024.10.20.1)

What's Changed

• moving to calendar versioning (#453) @devsetgo
• pip(deps): bump tox from 4.21.0 to 4.23.0 (#452) @dependabot
• pip(deps): bump fastapi[all] from 0.114.2 to 0.115.0 (#451) @dependabot
• pip(deps): bump tox from 4.18.1 to 4.21.0 (#450) @dependabot
• pip(deps): bump watchdog from 5.0.2 to 5.0.3 (#449) @dependabot
• pip(deps): bump pylint from 3.2.7 to 3.3.1 (#448) @dependabot
• pip(deps): bump ruff from 0.6.5 to 0.6.8 (#447) @dependabot

Published Date: 2024 October 20, 16:30

Complete Replacement of CX-Oracle for OracleDB (v0.14.4)

What's Changed

• Remove CX-Oracle for OracleDB cleanup (#446) @devsetgo
• pip(deps): bump pylint from 3.2.6 to 3.2.7 (#442) @dependabot
• pip(deps): bump mkdocs-material from 9.5.33 to 9.5.34 (#443) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#444) @dependabot
• github actionts(deps): bump actions/setup-python from 2 to 5 (#445) @dependabot

Published Date: 2024 September 15, 15:28

Standard Logging Suppression by Default (v0.14.3)

What's Changed

• Limit Standard Logging being Displayed (#441) @devsetgo

Published Date: 2024 August 31, 17:33

Improvements and fixes (v0.14.2)

What's Changed

• Improvements and fixes (#440) @devsetgo

Breaking changes

• save_text function no longer adds .txt by default.
• Change from cx-oracle to oracledb
• Improvements to documentation

Published Date: 2024 August 31, 00:02

Adding DB Disconnect (v0.14.1)

What's Changed

• Adding Database Disconnect (#439) @devsetgo
• pip(deps): bump pre-commit from 3.7.1 to 3.8.0 (#434) @dependabot
• updates to deal with stashing pages (#437) @devsetgo
• working on issue for deployment (#436) @devsetgo
• Adding MKDocs Workflow (#435) @devsetgo
• Version 0.14.0 (#433) @devsetgo

Published Date: 2024 August 25, 18:47

Fix of version for Pypi (v0.14.0-a)

What's Changed

• Version 0.14.0 (#433) @devsetgo

Published Date: 2024 July 27, 22:40

High Speed Multi-Processing Improvements (v0.14.0)

What's Changed

• High Speed Logging for Loguru Multi-Processing (#432) @devsetgo
• Resilient Sink Fixes (#431) @devsetgo
• Fix of bug in resilient sink (#430) @devsetgo
• Adding Resiliency to Logging Config (#429) @devsetgo
• pip(deps): bump mkdocs-print-site-plugin from 2.4.1 to 2.5.0 (#422) @dependabot
• pip(deps): bump ruff from 0.4.5 to 0.4.7 (#420) @dependabot
• pip(deps): bump autopep8 from 2.1.1 to 2.2.0 (#421) @dependabot
• pip(deps): bump mkdocs-material from 9.5.24 to 9.5.25 (#423) @dependabot

Published Date: 2024 July 27, 22:28

(v0.13.0-republish)

What's Changed

Republishing v0.13.0 for pypi.

Published Date: 2024 May 26, 17:13

(v0.13.0)

What's Changed

• Breaking Change: Removing Limit and Offset from read queries (#419) @devsetgo

Published Date: 2024 May 26, 15:44

Adding missing requirement (v0.12.4)

What's Changed

• adding missing requirement (#417) @devsetgo

Published Date: 2024 May 16, 14:40

Adding Email Validation (v0.12.3)

What's Changed

• bump to 0.12.3 (#416) @devsetgo
• Add email validation capabilities (#415) @devsetgo
• pip(deps): bump mkdocs-material from 9.5.20 to 9.5.21 (#414) @dependabot
• pip(deps): bump ruff from 0.4.2 to 0.4.4 (#413) @dependabot
• pip(deps): bump coverage-badge from 1.1.0 to 1.1.1 (#409) @dependabot
• pip(deps): bump mkdocs-material from 9.5.18 to 9.5.20 (#408) @dependabot
• pip(deps): bump mkdocstrings[python,shell] from 0.24.3 to 0.25.0 (#407) @dependabot
• pip(deps): bump ruff from 0.4.1 to 0.4.2 (#410) @dependabot
• pip(deps): bump tqdm from 4.66.2 to 4.66.3 (#412) @dependabot

Published Date: 2024 May 16, 14:19

logging changes (v0.12.2)

What's Changed

• Logging Changes, Documentation Updates, Using Ruff (#406) @devsetgo

Published Date: 2024 April 22, 16:17

Updates for MetaData and All HTTP Codes (v0.12.1)

What's Changed

• Enhancements and fixes (#405) @devsetgo

Published Date: 2024 April 19, 18:52

Breaking Change: Base Schema per Database Type (v0.12.0)

What's Changed

• Adding new base schema for database types (#402) @devsetgo
• creating main release for 0.11.2 (#390) @devsetgo
• Working on bug in read_query (#389) @devsetgo
• Reorganizing Library Structure (#388) @devsetgo
• Python Build and Publish fix (#382) @devsetgo

Breaking Change - SchemaBase is now SchemaBaseSQLite

Published Date: 2024 April 13, 22:55

Reorganizing Library Stucture (v0.11.2-main)

What's Changed

• creating main release for 0.11.2 (#390) @devsetgo
• Working on bug in read_query (#389) @devsetgo
• Reorganizing Library Structure (#388) @devsetgo
• Python Build and Publish fix (#382) @devsetgo

Breaking Changes are included in this release for import paths. See documents for more information.

Published Date: 2024 February 17, 19:09

Read Query Fix Beta Testing (v0.11.2-beta1)

What's Changed

• Working on bug in read_query (#389) @devsetgo
• Reorganizing Library Structure (#388) @devsetgo
• Python Build and Publish fix (#382) @devsetgo

Published Date: 2024 February 16, 22:01

Pre-Release to test new structure and publishing (v0.11.2-beta)

What's Changed

• Reorganizing Library Structure (#388) @devsetgo
• Python Build and Publish fix (#382) @devsetgo

Published Date: 2024 February 10, 21:16

(v0.11.2-fix2)

What's Changed

• Python Build and Publish fix (#382) @devsetgo

Published Date: 2024 January 21, 15:01

Adding Delete Many and minor fixes (v0.11.2)

What's Changed

• Adding Delete Many and Other Updates (#381) @devsetgo
• pip(deps): bump mkdocs-material from 9.5.2 to 9.5.3 (#377) @dependabot
• pip(deps): bump fastapi[all] from 0.105.0 to 0.108.0 (#375) @dependabot
• pip(deps): bump sqlalchemy from 2.0.23 to 2.0.24 (#374) @dependabot
• pip(deps): bump pytest from 7.4.3 to 7.4.4 (#373) @dependabot
• pip(deps): bump black from 23.12.0 to 23.12.1 (#376) @dependabot
• github actionts(deps): bump actions/setup-python from 4 to 5 (#378) @dependabot

Published Date: 2024 January 20, 00:07

Breaking Change (v0.11.1)

What's Changed

• Bump of Version to 0.11.1 (#371) @devsetgo
• Query Improvement (#370) @devsetgo
• 368 get one record should return an empty value when called (#369) @devsetgo
• updating docs from v0.11.0 release (#367) @devsetgo

Published Date: 2023 December 23, 10:49

Full Release of New Features (v0.11.0)

What's Changed

• Prep for Release (#366) @devsetgo
• Fixing sonar settings (#365) @devsetgo
• Fixes and improvements (#364) @devsetgo
• Dev (#362) @devsetgo
• Fix of issues from Beta release (#361) @devsetgo
• 359 tables are created before create tables is called (#360) @devsetgo
• Change Log (#358) @devsetgo
• fixing latest-changes (#357) @devsetgo
• removing jinja template from Latest Changes Action (#356) @devsetgo
• Action fixing adding main (#355) @devsetgo
• Fixing actions (#354) @devsetgo
• Fixing Beta Publishing issues and Documentation Improvements (#353) @devsetgo
• Update setup.py for sub packages (#352) @devsetgo
• Import Bug Fix (#351) @devsetgo
• Latest Changes Action Fix (#350) @devsetgo
• Next Release (#349) @devsetgo
• Dev (#348) @devsetgo
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 17, 22:00

Beta Release with fixes for multiple issues (v0.11.0-beta3-fix1)

What's Changed

• Dev (#362) @devsetgo
• Fix of issues from Beta release (#361) @devsetgo
• 359 tables are created before create tables is called (#360) @devsetgo
• Change Log (#358) @devsetgo
• fixing latest-changes (#357) @devsetgo
• removing jinja template from Latest Changes Action (#356) @devsetgo
• Action fixing adding main (#355) @devsetgo
• Fixing actions (#354) @devsetgo
• Fixing Beta Publishing issues and Documentation Improvements (#353) @devsetgo
• Update setup.py for sub packages (#352) @devsetgo
• Import Bug Fix (#351) @devsetgo
• Latest Changes Action Fix (#350) @devsetgo
• Next Release (#349) @devsetgo
• Dev (#348) @devsetgo
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 17, 16:23

Fixing AsyncDatabase create tables (v0.11.0-beta3)

What's Changed

• Fix of issues from Beta release (#361) @devsetgo
• 359 tables are created before create tables is called (#360) @devsetgo
• Change Log (#358) @devsetgo
• fixing latest-changes (#357) @devsetgo
• removing jinja template from Latest Changes Action (#356) @devsetgo
• Action fixing adding main (#355) @devsetgo
• Fixing actions (#354) @devsetgo
• Fixing Beta Publishing issues and Documentation Improvements (#353) @devsetgo
• Update setup.py for sub packages (#352) @devsetgo
• Import Bug Fix (#351) @devsetgo
• Latest Changes Action Fix (#350) @devsetgo
• Next Release (#349) @devsetgo
• Dev (#348) @devsetgo
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 17, 16:18

Build Updates (v0.11.0-beta2)

What's Changed

• Change Log (#358) @devsetgo
• fixing latest-changes (#357) @devsetgo
• removing jinja template from Latest Changes Action (#356) @devsetgo
• Action fixing adding main (#355) @devsetgo
• Fixing actions (#354) @devsetgo
• Fixing Beta Publishing issues and Documentation Improvements (#353) @devsetgo
• Update setup.py for sub packages (#352) @devsetgo
• Import Bug Fix (#351) @devsetgo
• Latest Changes Action Fix (#350) @devsetgo
• Next Release (#349) @devsetgo
• Dev (#348) @devsetgo
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 16, 20:34

Beta Release with fixes for multiple issues (v0.11.0-beta1-fix5)

What's Changed

• fixing latest-changes (#357) @devsetgo
• removing jinja template from Latest Changes Action (#356) @devsetgo
• Action fixing adding main (#355) @devsetgo
• Fixing actions (#354) @devsetgo
• Fixing Beta Publishing issues and Documentation Improvements (#353) @devsetgo
• Update setup.py for sub packages (#352) @devsetgo
• Import Bug Fix (#351) @devsetgo
• Latest Changes Action Fix (#350) @devsetgo
• Next Release (#349) @devsetgo
• Dev (#348) @devsetgo
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 16, 16:33

Build Fixes (v0.11.0-beta1-fix4)

What's Changed

• Update setup.py for sub packages (#352) @devsetgo
• Import Bug Fix (#351) @devsetgo
• Latest Changes Action Fix (#350) @devsetgo
• Next Release (#349) @devsetgo
• Dev (#348) @devsetgo
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 12, 11:45

Async Database and FastAPI functions (v0.11.0-beta0)

What's Changed

• Dev (#348) @devsetgo - New functionality and documentation for FastAPI Endpoints and Async Database Functionality
• pip(deps): bump autopep8 from 2.0.2 to 2.0.4 (#343) @dependabot
• pip(deps): bump wheel from 0.41.2 to 0.42.0 (#345) @dependabot
• pip(deps): bump mkdocstrings[python] from 0.21.2 to 0.24.0 (#346) @dependabot
• pip(deps): bump mkdocs from 1.4.3 to 1.5.3 (#347) @dependabot
• pip(deps): bump flake8 from 6.0.0 to 6.1.0 (#332) @dependabot
• pip(deps): bump click from 8.1.3 to 8.1.7 (#337) @dependabot
• pip(deps): bump wheel from 0.40.0 to 0.41.2 (#339) @dependabot
• github actionts(deps): bump actions/checkout from 2 to 4 (#340) @dependabot
• pip(deps): bump mkdocs-material from 9.1.17 to 9.4.2 (#341) @dependabot
• pip(deps): bump black from 23.3.0 to 23.9.1 (#342) @dependabot
• pip(deps): bump mkdocs-material from 9.1.15 to 9.1.17 (#326) @dependabot
• pip(deps): bump pytest from 7.3.1 to 7.4.0 (#327) @dependabot
• pip(deps): bump mkdocs from 1.4.2 to 1.4.3 (#328) @dependabot
• pip(deps): bump autoflake from 2.1.1 to 2.2.0 (#329) @dependabot
• pip(deps): bump pre-commit from 3.2.2 to 3.3.3 (#330) @dependabot
• pip(deps): bump mkdocs-material from 9.1.9 to 9.1.15 (#325) @dependabot
• pip(deps): bump autoflake from 2.0.2 to 2.1.1 (#324) @dependabot
• pip(deps): bump pytest-xdist from 3.2.1 to 3.3.1 (#323) @dependabot
• pip(deps): bump tox from 4.4.11 to 4.5.2 (#322) @dependabot
• pip(deps): bump pytest-cov from 4.0.0 to 4.1.0 (#321) @dependabot
• pip(deps): bump loguru from 0.6.0 to 0.7.0 (#317) @dependabot
• pip(deps): bump mkdocs-gen-files from 0.4.0 to 0.5.0 (#314) @dependabot
• pip(deps): bump pylint from 2.17.2 to 2.17.4 (#319) @dependabot
• pip(deps): bump mkdocs-material from 9.1.6 to 9.1.9 (#320) @dependabot
• pip(deps): bump pytest from 7.3.0 to 7.3.1 (#318) @dependabot

Published Date: 2023 December 10, 20:17

Pattern Analysis Update and Bug Fix (v0.10.1)

What's Changed

• Improvement to the patterns analysis (#313) @devsetgo
• pip(deps): bump mkdocs-material from 9.1.3 to 9.1.5 (#308) @dependabot
• pip(deps): bump pre-commit from 3.2.0 to 3.2.1 (#310) @dependabot
• pip(deps): bump watchdog from 2.3.1 to 3.0.0 (#309) @dependabot
• pip(deps): bump pylint from 2.17.0 to 2.17.1 (#311) @dependabot
• pip(deps): bump tox from 4.4.7 to 4.4.8 (#312) @dependabot

Published Date: 2023 April 08, 21:45

ChatGPT Driven Improvements (v0.10.0)

ChatGPT

Using ChatGPT to improve tests, find bugs, and improve performance. Code coverage is at 100% and the code base appears to be performing better than before.

Major changes are in PR #304

What's Changed

• latest change fix for regex pattern. (#307) @devsetgo
• Dev (#306) @devsetgo
• Workflow changes (#305) @devsetgo
• ChatGPT Driven Improvements (#304) @devsetgo
• pip(deps): bump pre-commit from 3.0.2 to 3.1.1 (#300) @dependabot
• pip(deps): bump pytest-xdist from 3.1.0 to 3.2.0 (#302) @dependabot
• pip(deps): bump autoflake from 2.0.0 to 2.0.1 (#299) @dependabot
• pip(deps): bump watchdog from 2.1.9 to 2.3.1 (#301) @dependabot
• pip(deps): bump pytest from 7.2.0 to 7.2.1 (#303) @dependabot
• pip(deps): bump pylint from 2.15.7 to 2.16.1 (#298) @dependabot
• pip(deps): bump autopep8 from 2.0.0 to 2.0.1 (#289) @dependabot
• pip(deps): bump pylint from 2.15.7 to 2.15.10 (#295) @dependabot
• pip(deps): bump black from 22.10.0 to 23.1.0 (#294) @dependabot
• pip(deps): bump tox from 3.27.1 to 4.4.4 (#296) @dependabot
• pip(deps): bump pre-commit from 2.20.0 to 3.0.2 (#297) @dependabot

Published Date: 2023 April 01, 00:27