Database functions

File to manage the data base, connect, get a cursor on this connection.

connect(user, password, host, port, database)

Connect to a database.

Parameters:
  • user (str) – User used to log in to the database.

  • password (str) – Password of the user.

  • host (str) – Host used for the connection.

  • port (str) – Port used for the connection.

  • database (str) – Database to which we want to connect.

Returns:
  • Connection – A connection to a database if all is working.

Exceptions:
  • PSQLConnectionError – If we have an error connecting to the database.

Source code in py_linq_sql/utils/db.py
def connect(
    user: str,
    password: str,
    host: str,
    port: str,
    database: str,
) -> Connection:
    """
    Connect to a database.

    Args:
        user: User used to log in to the database.
        password: Password of the user.
        host: Host used for the connection.
        port: Port used for the connection.
        database: Database to which we want to connect.

    Returns:
        A connection to a database if all is working.

    Raises:
        PSQLConnectionError: If we have an error connecting to the database.
    """
    try:  # pragma: no cover
        connection = psycopg.connect(
            user=user,
            password=password,
            host=host,
            port=port,
            dbname=database,
        )
        return connection
    except (Exception, Error) as err:  # pragma: no cover
        raise PSQLConnectionError(f"Error while connecting to posgeSQL {err}") from err

get_cursor(connection)

Get a cursor on the given connection.

Parameters:
  • connection (Connection) – A connection to a database on which we want to execute commands.

Returns:
  • Cursor – A cursor on the given connection to a data base to execute commands.

Source code in py_linq_sql/utils/db.py
def get_cursor(connection: Connection) -> Cursor:
    """
    Get a cursor on the given connection.

    Args:
        connection: A connection to a database on which we want to execute commands.

    Returns:
        A cursor on the given connection to a data base to execute commands.
    """
    cursor = connection.cursor(row_factory=psycopg.rows.namedtuple_row)
    return cursor