Help for sql.close_connection

Project

sql

Function

close_connection

Sample CLI

gway sql close-connection

References

['info', 'warning']

Full Code

def close_connection(datafile=None, *, sql_engine="sqlite", all=False):
    """
    Explicitly close one or all cached database connections.
    """
    if all:
        for connection in _connection_cache.values():
            try:
                connection.close()
            except Exception as e:
                gw.warning(f"Failed to close connection: {e}")
        _connection_cache.clear()
        gw.info("All connections closed.")
        return

    key = (sql_engine, datafile or "default")
    connection = _connection_cache.pop(key, None)
    if connection:
        try:
            connection.close()
            gw.info(f"Closed connection: {key}")
        except Exception as e:
            gw.warning(f"Failed to close {key}: {e}")