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}")