feat: refactor database engine management with dual-backend support (OEP + local egon-data)

This commit introduces a comprehensive refactoring of how eDisGo manages
database connections, enabling seamless switching between the
OpenEnergyPlatform (OEP) and local egon-data PostgreSQL databases.

## Engine as EDisGo property (edisgo/edisgo.py)

- Add lazy-initialized `engine` property to the EDisGo class, replacing
  the previous pattern of passing engine explicitly to every function call
- Support multiple initialization modes via new constructor parameters:
  `engine`, `db_config_path`, `db_url`, `db_ssh`, `db_token`
- Default to OEP engine when no configuration is provided
- Add `__deepcopy__` method that excludes the unpicklable SQLAlchemy
  engine (contains _thread._local objects) and lets the copy lazily
  recreate its own connection

## Optional engine parameter in all DB functions

Make the `engine` parameter optional (default=None) in all 33 functions
across 8 modules that access the database. When engine is None, each
function falls back to `edisgo_object.engine`. This maintains full
backward compatibility — callers can still pass an explicit engine.

Modified modules:
- edisgo/io/timeseries_import.py (10 functions)
- edisgo/io/electromobility_import.py (4 functions)
- edisgo/io/dsm_import.py (3 functions)
- edisgo/io/heat_pump_import.py (2 functions)
- edisgo/io/generators_import.py (1 function)
- edisgo/io/storage_import.py (1 function)
- edisgo/network/heat.py (2 engine fallbacks)
- edisgo/network/timeseries.py (1 engine fallback)

## Local DB schema resolution (edisgo/tools/config.py)

Extend `import_tables_from_oep()` to handle local egon-data databases:
- Auto-resolve schema mismatches: tables may reside in different schemas
  locally vs on OEP (e.g. egon_etrago_bus is in "grid" locally but
  imported via "supply" on OEP). The method now searches all schemas
  when a table is not found in the expected schema.
- Handle tables without primary keys (e.g. egon_map_zensus_grid_districts,
  egon_daily_heat_demand_per_climate_zone) by passing all columns as
  synthetic PK via `__mapper_args__["primary_key"]`.

## SSH tunnel lifecycle management (edisgo/io/db.py)

- `ssh_tunnel()` now returns `tuple[str, SSHTunnelForwarder]` instead of
  just the port string, so the server object is no longer lost
- `engine()` stores the SSH server as `engine._ssh_server` for later
  cleanup via `server.stop()`
- This fixes the "I/O operation on closed file" logging errors caused by
  orphaned paramiko keepalive threads writing to closed log handlers

## Test infrastructure (tests/conftest.py)

- Add `--runlocal` flag to run all DB tests against both OEP and local
  egon-data database
- Add `--egon-data-config` flag for custom YAML config path
- Parametrize tests via `db_engine` fixture: each DB test runs as
  `test_name[oep]` and `test_name[local]` when --runlocal is active
- Add `pytest_sessionfinish` hook that disposes engines and stops SSH
  tunnels cleanly after all tests complete
- Suppress paramiko DEBUG keepalive logging (defense-in-depth)
- Migrate all 29 test methods across 10 test files from hardcoded
  `pytest.engine` to parametrized `db_engine` fixture

## New files

- tests/io/test_db.py: 4 tests for SSH tunnel lifecycle (tunnel returns
  server, engine stores server, cleanup stops tunnel, OEP has no tunnel)
- egon-data.configuration.yaml: Template config file with placeholder
  credentials for local egon-data database connections

Author: Jonas Danke

.gitignore

@@ -28,3 +28,5 @@ eDisGo.egg-info/
 .vscode/settings.json
 
 *OEP_TOKEN.*
+egon-data.configuration.yaml
+.coverage

edisgo/edisgo.py

@@ -31,7 +31,6 @@
     pypsa_io,
     timeseries_import,
 )
-from edisgo.io.db import engine as egon_engine
 from edisgo.io.ding0_import import import_ding0_grid
 from edisgo.io.electromobility_import import (
     distribute_charging_demand,
@@ -131,6 +130,26 @@ class EDisGo:
         Default: "default".
     legacy_ding0_grids : bool
         Allow import of old ding0 grids. Default: True.
+    engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`, optional
+        SQLAlchemy database engine. If provided, this engine is used for all
+        database queries. If not provided, an engine is lazily created based
+        on other database parameters or defaults to an OEP connection.
+        Default: None.
+    db_config_path : str or pathlib.Path, optional
+        Path to an egon-data YAML configuration file (e.g.
+        ``~/.ssh/egon-data.configuration.yaml``). Used together with `db_ssh`
+        to create a local database engine via SSH tunnel.
+        Default: None.
+    db_url : str, optional
+        SQLAlchemy database URL for direct connection (e.g.
+        ``postgresql+psycopg2://egon:data@localhost:59510/egon-data``).
+        Default: None.
+    db_ssh : bool
+        If True, use SSH tunnel when connecting via `db_config_path`.
+        Default: False.
+    db_token : str or pathlib.Path, optional
+        OEP token or path to token file for OEP connection.
+        Default: None.
 
     Attributes
     ----------
@@ -157,13 +176,50 @@ class EDisGo:
         requirements or power plant dispatch.
     dsm : :class:`~.network.dsm.DSM`
         This is a container holding data on demand side management potential.
+    engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`
+        Database engine for OEDB access. Lazily initialized on first access.
+        Defaults to OEP connection. Can be overridden via setter or constructor
+        parameters.
+
+    Examples
+    --------
+    Default OEP connection (engine created lazily on first DB access):
+
+    >>> edisgo = EDisGo(ding0_grid="path/to/grid")
+    >>> edisgo.import_generators(generator_scenario="eGon2035")
+
+    Local database via YAML config with SSH tunnel:
+
+    >>> edisgo = EDisGo(
+    ...     ding0_grid="path/to/grid",
+    ...     db_config_path="~/.ssh/egon-data.configuration.yaml",
+    ...     db_ssh=True,
+    ... )
+
+    Local database via connection string:
+
+    >>> edisgo = EDisGo(
+    ...     ding0_grid="path/to/grid",
+    ...     db_url="postgresql+psycopg2://egon:data@localhost:59510/egon-data",
+    ... )
+
+    Override engine after construction:
+
+    >>> edisgo.engine = my_custom_engine
 
     """
 
     def __init__(self, **kwargs):
         # load configuration
         self._config = Config(**kwargs)
 
+        # database engine configuration (lazy initialization)
+        self._engine = kwargs.get("engine", None)
+        self._db_config_path = kwargs.get("db_config_path", None)
+        self._db_url = kwargs.get("db_url", None)
+        self._db_ssh = kwargs.get("db_ssh", False)
+        self._db_token = kwargs.get("db_token", None)
+
         # instantiate topology object and load grid data
         self.topology = Topology(config=self.config)
         self.import_ding0_grid(
@@ -232,6 +288,66 @@ def config(self):
     def config(self, kwargs):
         self._config = Config(**kwargs)
 
+    @property
+    def engine(self):
+        """
+        Database engine for accessing the OEDB or local egon-data database.
+
+        Lazy initialization: the engine is only created when first accessed.
+        By default, creates an OEP engine using oedialect. Can be overridden
+        by providing a custom engine via the setter, or by passing `db_config_path`,
+        `db_url`, or `engine` to the EDisGo constructor.
+
+        Parameters
+        ----------
+        engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`
+            SQLAlchemy engine to use for database connections.
+
+        Returns
+        -------
+        :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`
+            Database engine.
+
+        """
+        if self._engine is None:
+            if self._db_url is not None:
+                from sqlalchemy import create_engine
+
+                self._engine = create_engine(self._db_url, echo=False)
+            elif self._db_config_path is not None:
+                from edisgo.io.db import engine as create_db_engine
+
+                self._engine = create_db_engine(
+                    path=self._db_config_path, ssh=self._db_ssh
+                )
+            else:
+                from edisgo.io.db import engine as create_db_engine
+
+                self._engine = create_db_engine(token=self._db_token)
+        return self._engine
+
+    @engine.setter
+    def engine(self, value):
+        self._engine = value
+
+    def __deepcopy__(self, memo):
+        import copy as _copy
+
+        # Engine (SQLAlchemy) contains _thread._local objects that cannot be
+        # pickled.  Exclude it from the deep copy and let the copy lazily
+        # recreate its own engine from the stored configuration parameters
+        # (_db_config_path, _db_url, _db_token etc. are all plain strings/bools
+        # and will be deep-copied normally).
+        cls = self.__class__
+        result = cls.__new__(cls)
+        memo[id(self)] = result
+        for k, v in self.__dict__.items():
+            if k == "_engine":
+                setattr(result, k, None)
+            else:
+                setattr(result, k, _copy.deepcopy(v, memo))
+        return result
+
     def import_ding0_grid(self, path, legacy_ding0_grids=True):
         """
         Import ding0 topology data from csv files in the format as
@@ -559,7 +675,7 @@ def set_time_series_active_power_predefined(
             is indexed using a default year and set for the whole year.
 
         """
-        engine = kwargs["engine"] if "engine" in kwargs else egon_engine()
+        engine = kwargs.pop("engine", None) or self.engine
         if self.timeseries.timeindex.empty:
             logger.warning(
                 "When setting time series using predefined profiles it is better to "
@@ -982,7 +1098,7 @@ def import_generators(self, generator_scenario=None, **kwargs):
             keyword arguments.
 
         """
-        engine = kwargs["engine"] if "engine" in kwargs else egon_engine()
+        engine = kwargs.pop("engine", None) or self.engine
         if self.legacy_grids is True:
             generators_import.oedb_legacy(
                 edisgo_object=self, generator_scenario=generator_scenario, **kwargs
@@ -1130,12 +1246,14 @@ def _check_convergence():
             if raise_not_converged and len(timesteps_not_converged) > 0:
                 raise ValueError(
                     "Power flow analysis did not converge for the "
-                    f"following {len(timesteps_not_converged)} time steps: {timesteps_not_converged}."
+                    f"following {len(timesteps_not_converged)} time "
+                    f"steps: {timesteps_not_converged}."
                 )
             elif len(timesteps_not_converged) > 0:
                 logger.warning(
                     "Power flow analysis did not converge for the "
-                    f"following {len(timesteps_not_converged)} time steps: {timesteps_not_converged}."
+                    f"following {len(timesteps_not_converged)} time "
+                    f"steps: {timesteps_not_converged}."
                 )
             return timesteps_converged, timesteps_not_converged
 
@@ -2032,6 +2150,9 @@ def import_electromobility(
         if import_electromobility_data_kwds is None:
             import_electromobility_data_kwds = {}
 
+        if engine is None:
+            engine = self.engine
+
         if data_source == "oedb":
             import_electromobility_from_oedb(
                 self,
@@ -2136,7 +2257,9 @@ def apply_charging_strategy(
             self, strategy=strategy, charging_park_ids=charging_park_ids, **kwargs
         )
 
-    def import_heat_pumps(self, scenario, engine, timeindex=None, import_types=None):
+    def import_heat_pumps(
+        self, scenario, engine=None, timeindex=None, import_types=None
+    ):
         """
         Gets heat pump data for specified scenario from oedb and integrates the heat
         pumps into the grid.
@@ -2212,6 +2335,9 @@ def import_heat_pumps(self, scenario, engine, timeindex=None, import_types=None)
             "central_resistive_heaters". If None, all are imported.
 
         """
+        if engine is None:
+            engine = self.engine
+
         # set up year to index data by
         # first try to get index from time index
         if timeindex is None:
@@ -2308,7 +2434,7 @@ def apply_heat_pump_operating_strategy(
         """
         hp_operating_strategy(self, strategy=strategy, heat_pump_names=heat_pump_names)
 
-    def import_dsm(self, scenario: str, engine: Engine, timeindex=None):
+    def import_dsm(self, scenario: str, engine: Engine = None, timeindex=None):
         """
         Gets industrial and CTS DSM profiles from the
         `OpenEnergy DataBase <https://openenergyplatform.org/database/>`_.
@@ -2327,8 +2453,8 @@ def import_dsm(self, scenario: str, engine: Engine, timeindex=None):
         scenario : str
             Scenario for which to retrieve DSM data. Possible options
             are 'eGon2035' and 'eGon100RE'.
-        engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`
-            Database engine.
+        engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`, optional
+            Database engine. If not provided, uses :attr:`~.EDisGo.engine`.
         timeindex : :pandas:`pandas.DatetimeIndex<DatetimeIndex>` or None
             Specifies time steps for which to get data. Leap years can currently not be
             handled. In case the given timeindex contains a leap year, the data will be
@@ -2340,6 +2466,9 @@ def import_dsm(self, scenario: str, engine: Engine, timeindex=None):
             is indexed using the default year and returned for the whole year.
 
         """
+        if engine is None:
+            engine = self.engine
+
         dsm_profiles = dsm_import.oedb(
             edisgo_obj=self, scenario=scenario, engine=engine, timeindex=timeindex
         )
@@ -2351,7 +2480,7 @@ def import_dsm(self, scenario: str, engine: Engine, timeindex=None):
     def import_home_batteries(
         self,
         scenario: str,
-        engine: Engine,
+        engine: Engine = None,
     ):
         """
         Gets home battery data for specified scenario and integrates the batteries into
@@ -2379,10 +2508,13 @@ def import_home_batteries(
         scenario : str
             Scenario for which to retrieve home battery data. Possible options
             are 'eGon2035' and 'eGon100RE'.
-        engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`
-            Database engine.
+        engine : :sqlalchemy:`sqlalchemy.Engine<sqlalchemy.engine.Engine>`, optional
+            Database engine. If not provided, uses :attr:`~.EDisGo.engine`.
 
         """
+        if engine is None:
+            engine = self.engine
+
         home_batteries_oedb(
             edisgo_obj=self,
             scenario=scenario,
@@ -3640,8 +3772,10 @@ def _check_timeindex(check_df, param_name):
             ).any()
             if comparison.any():
                 logger.warning(
-                    "Heat demand is higher than rated heatpump power"
-                    f" of heatpumps: {comparison.index[comparison.values].values}. Demand can not be covered if no sufficient"
+                    "Heat demand is higher than rated heatpump "
+                    "power of heatpumps: "
+                    f"{comparison.index[comparison.values].values}"
+                    ". Demand can not be covered if no sufficient"
                     " heat storage capacities are available."
                 )