SqlContent

Class to handle SQL content preparation and execution.

Source code in pum/sql_content.py

class SqlContent:
    """Class to handle SQL content preparation and execution."""

    def __init__(self, sql: str | psycopg.sql.SQL | Path) -> None:
        """Initialize the SqlContent class.

        Args:
            sql: The SQL statement to execute or a path to a SQL file.

        """
        self.sql = sql

    def validate(self, parameters: dict | None) -> bool:
        """Validate the SQL content.
        This is done by checking if the SQL content is not empty.

        Args:
            parameters: The parameters to pass to the SQL files.

        Returns:
            bool: True if valid, False otherwise.

        """
        if not self.sql:
            raise PumSqlError("SQL content is empty.")
        self._prepare_sql(parameters)
        return True

    def execute(
        self,
        connection: psycopg.Connection,
        *,
        parameters: dict | None = None,
        commit: bool = False,
    ) -> psycopg.Cursor:
        """Execute a SQL statement with optional parameters.

        Args:
            connection: The database connection to execute the SQL statement.
            parameters: Parameters to bind to the SQL statement. Defaults to ().
            commit: Whether to commit the transaction. Defaults to False.

        """
        cursor = connection.cursor()

        for sql_code in self._prepare_sql(parameters):
            try:
                statement = sql_code.as_string(connection)
            except (psycopg.errors.SyntaxError, psycopg.errors.ProgrammingError) as e:
                raise PumSqlError(
                    f"SQL preparation failed for the following code: {statement} {e}"
                ) from e
            try:
                logger.debug(f"Executing SQL statement: {statement}")
                cursor.execute(statement)
            except (psycopg.errors.SyntaxError, psycopg.errors.ProgrammingError) as e:
                raise PumSqlError(
                    f"SQL execution failed for the following code: {statement} {e}"
                ) from e
        if commit:
            connection.commit()

        return cursor

    def _prepare_sql(self, parameters: dict | None) -> list[psycopg.sql.SQL]:
        """Prepare SQL for execution.

        Args:
            sql: The SQL statement to execute or a path to a SQL file.
            parameters: Parameters to bind to the SQL statement. Defaults to ().

        Returns:
            list: A list of prepared SQL statements.

        Raises:
            PumSqlError: If SQL preparation fails.

        """
        if isinstance(self.sql, Path):
            logger.info(
                f"Checking SQL from file: {self.sql} with parameters: {parameters}",
            )
            sql_code = sql_chunks_from_file(self.sql)
        elif isinstance(self.sql, str):
            sql_code = [psycopg.sql.SQL(self.sql)]
        else:
            sql_code = [self.sql]

        def format_sql(
            statement: psycopg.sql.SQL, parameters: dict | None = None
        ) -> psycopg.sql.SQL:
            for key, value in (parameters or {}).items():
                if (
                    not isinstance(value, psycopg.sql.Literal)
                    and not isinstance(value, psycopg.sql.Identifier)
                    and not isinstance(value, psycopg.sql.Composed)
                ):
                    raise PumSqlError(
                        f"Invalid parameter type for key '{key}': {type(value)}. "
                        "Parameters must be psycopg.sql.Literal or psycopg.sql.Identifier."
                    )
            try:
                return statement.format(**parameters)
            except TypeError:
                # if parameters is None, we can ignore this error
                return statement
            except KeyError as e:
                raise PumSqlError(
                    f"SQL preparation failed for the following code: missing parameter: {statement} {e}"
                ) from e

        return [format_sql(statement, parameters) for statement in sql_code]

__init__

__init__(sql: str | SQL | Path) -> None

Initialize the SqlContent class.

Parameters:

Name	Type	Description	Default
sql	str | SQL | Path	The SQL statement to execute or a path to a SQL file.	required

Source code in pum/sql_content.py

def __init__(self, sql: str | psycopg.sql.SQL | Path) -> None:
    """Initialize the SqlContent class.

    Args:
        sql: The SQL statement to execute or a path to a SQL file.

    """
    self.sql = sql

execute

execute(connection: Connection, *, parameters: dict | None = None, commit: bool = False) -> psycopg.Cursor

Execute a SQL statement with optional parameters.

Parameters:

Name	Type	Description	Default
connection	Connection	The database connection to execute the SQL statement.	required
parameters	dict | None	Parameters to bind to the SQL statement. Defaults to ().	None
commit	bool	Whether to commit the transaction. Defaults to False.	False

Source code in pum/sql_content.py

def execute(
    self,
    connection: psycopg.Connection,
    *,
    parameters: dict | None = None,
    commit: bool = False,
) -> psycopg.Cursor:
    """Execute a SQL statement with optional parameters.

    Args:
        connection: The database connection to execute the SQL statement.
        parameters: Parameters to bind to the SQL statement. Defaults to ().
        commit: Whether to commit the transaction. Defaults to False.

    """
    cursor = connection.cursor()

    for sql_code in self._prepare_sql(parameters):
        try:
            statement = sql_code.as_string(connection)
        except (psycopg.errors.SyntaxError, psycopg.errors.ProgrammingError) as e:
            raise PumSqlError(
                f"SQL preparation failed for the following code: {statement} {e}"
            ) from e
        try:
            logger.debug(f"Executing SQL statement: {statement}")
            cursor.execute(statement)
        except (psycopg.errors.SyntaxError, psycopg.errors.ProgrammingError) as e:
            raise PumSqlError(
                f"SQL execution failed for the following code: {statement} {e}"
            ) from e
    if commit:
        connection.commit()

    return cursor

validate

validate(parameters: dict | None) -> bool

Validate the SQL content. This is done by checking if the SQL content is not empty.

Parameters:

Name	Type	Description	Default
parameters	dict | None	The parameters to pass to the SQL files.	required

Returns:

Name	Type	Description
bool	bool	True if valid, False otherwise.

Source code in pum/sql_content.py

def validate(self, parameters: dict | None) -> bool:
    """Validate the SQL content.
    This is done by checking if the SQL content is not empty.

    Args:
        parameters: The parameters to pass to the SQL files.

    Returns:
        bool: True if valid, False otherwise.

    """
    if not self.sql:
        raise PumSqlError("SQL content is empty.")
    self._prepare_sql(parameters)
    return True