Help for sql.close_connection

Sample CLI

gway sql close-connection

References

Full Code

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

    base_key = (sql_engine, datafile or "default")
    thread_key = threading.get_ident() if sql_engine == "sqlite" else "*"
    key = (base_key, thread_key)
    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}")