tests: update env.py and conftest so tests use unique databases (#3654)

* Refactor Alembic environment script to simplify database connection logic and remove unused imports

* Refactor test client fixture to use TemporaryDirectory and UUID for database path

* Add `reload_engine` method to reinitialize the database engine

* Add cleanup steps to test fixtures to ensure proper resource management

- Added cleanup steps to various pytest fixtures to ensure temporary directories, database entries, and other resources are properly cleaned up after tests.
- Modified fixtures to use `yield` for better resource management.
- Ensured database connections are closed and tables are dropped after tests.
- Improved temporary directory handling with context managers.

* Add unit test to verify return type of list_flows method in custom component

* Refactor tests in `test_custom_component.py` to remove unused imports and fixtures, and update existing fixtures for consistency.

* Add debug checks for user existence in `test_user_waiting_for_approval` test

* Fix import order and add flow_id validation in transaction handling

- Corrected the import order in `utils.py`.
- Added validation for `flow_id` in the transaction handling logic to ensure it is set correctly.

* Add function to delete transactions by flow ID in CRUD module

* Add cleanup for transactions and vertex builds in test teardown

- Introduced `_delete_transactions_and_vertex_builds` function to remove transactions and vertex builds associated with a user.
- Updated `user` fixture to call the new cleanup function before deleting the user.

* Refactor flow_id assignment logic in `utils.py` to improve readability and correctness

* [autofix.ci] apply automated fixes

* Refactor test to use pytest's tmp_path fixture for temporary directory creation

* Convert `test_user_waiting_for_approval` to an async test function

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

src/backend/base/langflow/alembic/env.py

@@ -1,9 +1,6 @@
-import os
-import warnings
 from logging.config import fileConfig
 
 from alembic import context
-from loguru import logger
 from sqlalchemy import engine_from_config, pool
 
 from langflow.services.database.models import *
@@ -42,8 +39,7 @@ def run_migrations_offline() -> None:
     script output.
 
     """
-    url = os.getenv("LANGFLOW_DATABASE_URL")
-    url = url or config.get_main_option("sqlalchemy.url")
+    url = config.get_main_option("sqlalchemy.url")
     context.configure(
         url=url,
         target_metadata=target_metadata,
@@ -63,32 +59,17 @@ def run_migrations_online() -> None:
     and associate a connection with the context.
 
     """
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
 
-    try:
-        from langflow.services.database.factory import DatabaseServiceFactory
-        from langflow.services.deps import get_db_service
-        from langflow.services.manager import initialize_settings_service, service_manager
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
 
-        initialize_settings_service()
-        service_manager.register_factory(DatabaseServiceFactory())
-        connectable = get_db_service().engine
-    except Exception:
-        logger.exception("Error getting database engine")
-        url = os.getenv("LANGFLOW_DATABASE_URL")
-        url = url or config.get_main_option("sqlalchemy.url")
-        if url:
-            config.set_main_option("sqlalchemy.url", url)
-        connectable = engine_from_config(
-            config.get_section(config.config_ini_section, {}),
-            prefix="sqlalchemy.",
-            poolclass=pool.NullPool,
-        )
-    with warnings.catch_warnings():
-        warnings.simplefilter("ignore")
-        with connectable.connect() as connection:
-            context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
-            with context.begin_transaction():
-                context.run_migrations()
+        with context.begin_transaction():
+            context.run_migrations()
 
 
 if context.is_offline_mode():

src/backend/base/langflow/graph/utils.py

@@ -141,6 +141,11 @@ async def log_transaction(
     try:
         if not get_settings_service().settings.transactions_storage_enabled:
             return
+        if not flow_id:
+            if source.graph.flow_id:
+                flow_id = source.graph.flow_id
+            else:
+                return
         inputs = _vertex_to_primitive_dict(source)
         transaction = TransactionBase(
             vertex_id=source.id,

src/backend/base/langflow/services/database/service.py

@@ -45,6 +45,9 @@ class DatabaseService(Service):
         self.alembic_cfg_path = langflow_dir / "alembic.ini"
         self.engine = self._create_engine()
 
+    def reload_engine(self):
+        self.engine = self._create_engine()
+
     def _create_engine(self) -> Engine:
         """Create the engine for the database."""
         if self.settings_service.settings.database_url and self.settings_service.settings.database_url.startswith(

src/backend/base/langflow/services/telemetry/opentelemetry.py

@@ -112,6 +112,8 @@ class ThreadSafeSingletonMetaUsingWeakref(type):
 
 class OpenTelemetry(metaclass=ThreadSafeSingletonMetaUsingWeakref):
     _metrics_registry: dict[str, Metric] = {}
+    _metrics: dict[str, Counter | ObservableGaugeWrapper | Histogram | UpDownCounter] = {}
+    _meter_provider: MeterProvider | None = None
 
     def _add_metric(self, name: str, description: str, unit: str, metric_type: MetricType, labels: dict[str, bool]):
         metric = Metric(name=name, description=description, metric_type=metric_type, unit=unit, labels=labels)
@@ -140,33 +142,37 @@ class OpenTelemetry(metaclass=ThreadSafeSingletonMetaUsingWeakref):
             labels={"flow_id": mandatory_label},
         )
 
-    _metrics: dict[str, Counter | ObservableGaugeWrapper | Histogram | UpDownCounter] = {}
-
     def __init__(self, prometheus_enabled: bool = True):
-        self._register_metric()
+        if not self._metrics_registry:
+            self._register_metric()
 
-        resource = Resource.create({"service.name": "langflow"})
-        metric_readers = []
+        if self._meter_provider is None:
+            resource = Resource.create({"service.name": "langflow"})
+            metric_readers = []
 
-        # configure prometheus exporter
-        self.prometheus_enabled = prometheus_enabled
-        if prometheus_enabled:
-            metric_readers.append(PrometheusMetricReader())
+            # configure prometheus exporter
+            self.prometheus_enabled = prometheus_enabled
+            if prometheus_enabled:
+                metric_readers.append(PrometheusMetricReader())
 
-        meter_provider = MeterProvider(resource=resource, metric_readers=metric_readers)
-        metrics.set_meter_provider(meter_provider)
-        self.meter = meter_provider.get_meter(langflow_meter_name)
+            self._meter_provider = MeterProvider(resource=resource, metric_readers=metric_readers)
+            metrics.set_meter_provider(self._meter_provider)
+
+        self.meter = self._meter_provider.get_meter(langflow_meter_name)
 
         for name, metric in self._metrics_registry.items():
             if name != metric.name:
                 msg = f"Key '{name}' does not match metric name '{metric.name}'"
                 raise ValueError(msg)
-            with warnings.catch_warnings():
-                warnings.simplefilter("ignore")
-
-                self._metrics[metric.name] = self._create_metric(metric)
+            if name not in self._metrics:
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore")
+                    self._metrics[metric.name] = self._create_metric(metric)
 
     def _create_metric(self, metric):
+        if metric.name in self._metrics:
+            return self._metrics[metric.name]
+
         if metric.type == MetricType.COUNTER:
             return self.meter.create_counter(
                 name=metric.name,