firebirdsql_run

Public accessible objects of that module.

main

Firebirdsql wrapper inspired by subprocess.run.

connection

connection(db: Path | str, host: str = '127.0.0.1', port: int = 3050, user: str = 'TWUSER', passwd: str = '', access: DBAccess = DBAccess.READ_WRITE) -> Connection

Create a connection to a Firebird database using the firebirdsql library.

Parameters:

Name	Type	Description	Default
db	Path | str	The path or name of the database.	required
host	str	The host address of the server.	'127.0.0.1'
port	int	The port number of the server.	3050
user	str	The username for authentication.	'TWUSER'
passwd	str	The password. If omitted, it is taken from FIREBIRD_KEY environment variable.	''
access	DBAccess	The access mode for the connection.	READ_WRITE

Returns:

Name	Type	Description
conn	Connection	The connection object.

Source code in src/firebirdsql_run/main.py

def connection(
    db: Path | str,
    host: str = "127.0.0.1",
    port: int = 3050,
    user: str = "TWUSER",
    passwd: str = "",
    access: DBAccess = DBAccess.READ_WRITE,
) -> Connection:
    """Create a connection to a Firebird database using the firebirdsql library.

    Args:
        db: The path or name of the database.
        host: The host address of the server.
        port: The port number of the server.
        user: The username for authentication.
        passwd: The password. If omitted, it is taken from `FIREBIRD_KEY` environment variable.
        access: The access mode for the connection.

    Returns:
        conn: The connection object.
    """
    return connect(
        host=getfqdn(host),
        database=f"{db}",
        port=port,
        user=user,
        password=passwd or get_env("FIREBIRD_KEY"),
        isolation_level=access.value,
    )

execute

execute(query: str, params: tuple = (), db: Path | str = '', host: str = '127.0.0.1', port: int = 3050, user: str = 'TWUSER', passwd: str = '', access: DBAccess = DBAccess.READ_WRITE, use_conn: Connection | None = None) -> CompletedTransaction

Execute a transaction in a Firebird database.

Parameters:

Name	Type	Description	Default
query	str	The SQL query to execute.	required
params	tuple	The parameters to be used in the query.	()
db	Path | str	The path or name of the database.	''
host	str	The host address of the server.	'127.0.0.1'
port	int	The port number of the server.	3050
user	str	The username for authentication.	'TWUSER'
passwd	str	The password. If omitted, it is taken from FIREBIRD_KEY environment variable.	''
access	DBAccess	The access mode for the connection.	READ_WRITE
use_conn	Connection | None	An existing connection to use. Takes precedence over the default connection settings.	None

Returns:

Name	Type	Description
result	CompletedTransaction	An named tuple containing the transaction details, including the query result.

Source code in src/firebirdsql_run/main.py

def execute(
    query: str,
    params: tuple = (),
    db: Path | str = "",
    host: str = "127.0.0.1",
    port: int = 3050,
    user: str = "TWUSER",
    passwd: str = "",
    access: DBAccess = DBAccess.READ_WRITE,
    use_conn: Connection | None = None,
) -> CompletedTransaction:
    """Execute a transaction in a Firebird database.

    Args:
        query: The SQL query to execute.
        params: The parameters to be used in the query.
        db: The path or name of the database.
        host: The host address of the server.
        port: The port number of the server.
        user: The username for authentication.
        passwd: The password. If omitted, it is taken from `FIREBIRD_KEY` environment variable.
        access: The access mode for the connection.
        use_conn: An existing connection to use. Takes precedence over the default connection settings.

    Returns:
        result: An named tuple containing the transaction details, including the query result.
    """
    conn: Connection | None = None
    timer: Timer | None = None

    try:
        conn = use_conn or connection(
            host=host,
            db=db,
            port=port,
            user=user,
            passwd=passwd,
            access=access,
        )
        with conn.cursor() as cur, Timer() as timer:
            cur.execute(query=query, params=params)
            lines = cur.fetchall()
            descr = cur.description

    except Exception as e:  # noqa: BLE001
        data = []
        returncode = 1
        exception = f"{e}"
    else:
        columns = [] if descr is None else [f"{col[0]}".lower() for col in descr]
        data = [] if lines is None else [dict(zip(columns, line, strict=True)) for line in lines]
        returncode = 0
        exception = ""
    finally:
        if conn is not None:
            conn.commit()
            if use_conn is None:
                conn.close()

    return CompletedTransaction(
        host=host if conn is None else conn.hostname,
        port=port if conn is None else conn.port,
        db=f"{db}" if conn is None else f"{conn.filename}",
        user=user if conn is None else f"{conn.user}",
        access=access.name if conn is None else DBAccess(conn.isolation_level).name,
        returncode=returncode,
        exception=exception,
        query=query,
        params=params,
        time=0 if timer is None else round(timer.interval, 5),
        data=data,
    )

make_query

make_query(procname: str, params: tuple) -> str

Create a query for a stored procedure in a Firebird database.

Parameters:

Name	Type	Description	Default
procname	str	The name of the stored procedure to execute.	required
params	tuple	The parameters to pass to the stored procedure.	required

Returns:

Name	Type	Description
query	str	The query for the stored procedure.

Source code in src/firebirdsql_run/main.py

def make_query(procname: str, params: tuple) -> str:
    """Create a query for a stored procedure in a Firebird database.

    Args:
        procname: The name of the stored procedure to execute.
        params: The parameters to pass to the stored procedure.

    Returns:
        query: The query for the stored procedure.
    """
    return f"EXECUTE PROCEDURE {procname} " + ",".join("?" * len(params))

callproc

callproc(procname: str, params: tuple = (), db: Path | str = '', host: str = '127.0.0.1', port: int = 3050, user: str = 'TWUSER', passwd: str = '', access: DBAccess = DBAccess.READ_WRITE, use_conn: Connection | None = None) -> CompletedTransaction

Execute a stored procedure in a Firebird database.

Parameters:

Name	Type	Description	Default
procname	str	The name of the stored procedure to execute.	required
params	tuple	The parameters to pass to the stored procedure.	()
db	Path | str	The path or name of the database.	''
host	str	The host address of the server.	'127.0.0.1'
port	int	The port number of the server.	3050
user	str	The username for authentication.	'TWUSER'
passwd	str	The password. If omitted, it is taken from FIREBIRD_KEY environment variable.	''
access	DBAccess	The access mode for the connection.	READ_WRITE
use_conn	Connection | None	An existing connection to use. Takes precedence over the default connection settings.	None

Returns:

Name	Type	Description
result	CompletedTransaction	An named tuple containing the transaction details, including the query result.

Source code in src/firebirdsql_run/main.py

def callproc(
    procname: str,
    params: tuple = (),
    db: Path | str = "",
    host: str = "127.0.0.1",
    port: int = 3050,
    user: str = "TWUSER",
    passwd: str = "",
    access: DBAccess = DBAccess.READ_WRITE,
    use_conn: Connection | None = None,
) -> CompletedTransaction:
    """Execute a stored procedure in a Firebird database.

    Args:
        procname: The name of the stored procedure to execute.
        params: The parameters to pass to the stored procedure.
        db: The path or name of the database.
        host: The host address of the server.
        port: The port number of the server.
        user: The username for authentication.
        passwd: The password. If omitted, it is taken from `FIREBIRD_KEY` environment variable.
        access: The access mode for the connection.
        use_conn: An existing connection to use. Takes precedence over the default connection settings.

    Returns:
        result: An named tuple containing the transaction details, including the query result.
    """
    return execute(
        query=make_query(procname, params),
        params=params,
        host=host,
        port=port,
        db=db,
        user=user,
        passwd=passwd,
        use_conn=use_conn,
        access=access,
    )

type

Type definitions for the package.

FBTypes module-attribute

FBTypes = str | float | datetime | None

Type alias for the possible types of data returned by a FirebirdSQL query.

Dataset module-attribute

Dataset = list[dict[str, FBTypes]]

Type alias for a list of dictionaries representing a dataset.

DBAccess

Bases: IntEnum

Enumeration of access modes for FirebirdSQL connections.

Attributes:

Name	Type	Description
READ_ONLY		Read-only access mode.
READ_WRITE		Read-write access mode.

CompletedTransaction

Bases: NamedTuple

Represents a completed transaction in a database.

Attributes:

Name	Type	Description
host	str	The host address of the server.
port	int	The port number of the server.
db	str	The database where the transaction was executed.
user	str	The user who executed the transaction.
access	str	The access mode used for the transaction.
returncode	int	The return code of the transaction execution.
exception	str	The exception message if the transaction failed.
query	str	The SQL query executed in the transaction.
params	tuple	The parameters used in the SQL query.
time	float	The number of seconds it took to execute the transaction.
data	Dataset	The data returned by the transaction, represented as a list of dictionaries.

ExecuteError

Bases: Exception

Exception raised when an error occurs during the transaction execution.

util

Utility functions for the package.

GetEnvError

Bases: Exception

Exception raised when an environment variable is not found.

Timer

A context manager for measuring the execution time of a code block.

get_env

get_env(variable: str) -> str

Get the value of an environment variable.

Parameters:

Name	Type	Description	Default
variable	str	The name of the environment variable to retrieve.	required

Returns:

Name	Type	Description
value	str	The value of the environment variable, if found.

Raises:

Type	Description
GetEnvError	If the environment variable is not found.

Source code in src/firebirdsql_run/util.py

def get_env(variable: str) -> str:
    """Get the value of an environment variable.

    Args:
        variable: The name of the environment variable to retrieve.

    Returns:
        value: The value of the environment variable, if found.

    Raises:
        GetEnvError: If the environment variable is not found.
    """
    if value := getenv(variable):
        return value

    msg = f"Please setup {variable} in environment variables."
    raise GetEnvError(msg)