incendium.db.DisposableConnection

Class DisposableConnection

• object
  • incendium.db.DisposableConnection

Description

A disposable connection enables a database connection in Ignition and disables it once the operation is completed to release resources.

Syntax

DisposableConnection(database, [retries])

Args:

• database (str): The name of the database connection in Ignition.
• retries (int): The number of additional times to retry enabling the connection. Optional. Defaults to 3.

Returns:

• DisposableConnection: A DisposableConnection instance.

Recommendations

DisposableConnection is intended when a Database connection must be disabled immediately after usage.

Code Examples

Executing a query using system.db.

import traceback

import system.db
from incendium import constants, util
from incendium.db import DisposableConnection
from incendium.exceptions import ApplicationError
from java.lang import Exception as JavaException


def some_function():
    result = None
    try:
        with DisposableConnection("test") as conn:
            result = system.db.runQuery("SELECT * FROM [table]", conn.database)
    except IOError as exc:
        message = constants.UNEXPECTED_ERROR.format(
            util.get_function_name(),
            "\n".join(traceback.format_exc().splitlines()),
        )
        raise ApplicationError(message, exc)
    except JavaException as exc:
        message = constants.UNEXPECTED_ERROR_CAUSED_BY.format(
            util.get_function_name(),
            "\n".join(traceback.format_exc().splitlines()),
            exc.cause,
        )
        raise ApplicationError(message, exc, exc.cause)

    return result

Calling a stored procedure using incendium.db.

import traceback

import incendium.db
import system.db
from incendium import constants, util
from incendium.db import DisposableConnection, InParam
from incendium.exceptions import ApplicationError
from java.lang import Exception as JavaException


def some_function():
    try:
        with DisposableConnection("Some Connection", retries=10) as conn:
            params = [InParam("int_param", system.db.INTEGER, 1)]
            incendium.db.execute_non_query(
                "[schema].[procedure]", database=conn.database, params=params
            )
    except IOError as exc:
        message = constants.UNEXPECTED_ERROR.format(
            util.get_function_name(),
            "\n".join(traceback.format_exc().splitlines()),
        )
        raise ApplicationError(message, exc)
    except JavaException as exc:
        message = constants.UNEXPECTED_ERROR_CAUSED_BY.format(
            util.get_function_name(),
            "\n".join(traceback.format_exc().splitlines()),
            exc.cause,
        )
        raise ApplicationError(message, exc, exc.cause)