Fix sqlalchemy uninstrument

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fix SQLAlchemy uninstrumentation
+  ([#1581](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1581))
 - Fix aiopg instrumentation to work with aiopg < 2.0.0
   ([#1473](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1473))
 - `opentelemetry-instrumentation-aws-lambda` Adds an option to configure `disable_aws_context_propagation` by

instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/__init__.py

@@ -119,6 +119,8 @@ class SQLAlchemyInstrumentor(BaseInstrumentor):
     See `BaseInstrumentor`
     """
 
+    engines = []
+
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
@@ -158,17 +160,22 @@ def _instrument(self, **kwargs):
                 "create_async_engine",
                 _wrap_create_async_engine(tracer_provider, enable_commenter),
             )
+
+        self.engines = []
         if kwargs.get("engine") is not None:
-            return EngineTracer(
-                _get_tracer(tracer_provider),
-                kwargs.get("engine"),
-                kwargs.get("enable_commenter", False),
-                kwargs.get("commenter_options", {}),
+            self.engines.append(
+                EngineTracer(
+                    _get_tracer(tracer_provider),
+                    kwargs.get("engine"),
+                    kwargs.get("enable_commenter", False),
+                    kwargs.get("commenter_options", {}),
+                )
             )
+            return self.engines[0]
         if kwargs.get("engines") is not None and isinstance(
             kwargs.get("engines"), Sequence
         ):
-            return [
+            self.engines = [
                 EngineTracer(
                     _get_tracer(tracer_provider),
                     engine,
@@ -177,6 +184,7 @@ def _instrument(self, **kwargs):
                 )
                 for engine in kwargs.get("engines")
             ]
+            return self.engines
 
         return None
 
@@ -186,3 +194,5 @@ def _uninstrument(self, **kwargs):
         unwrap(Engine, "connect")
         if parse_version(sqlalchemy.__version__).release >= (1, 4):
             unwrap(sqlalchemy.ext.asyncio, "create_async_engine")
+        for engine in self.engines:
+            engine.remove_event_listeners()

instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py

@@ -14,7 +14,10 @@
 import os
 import re
 
-from sqlalchemy.event import listen  # pylint: disable=no-name-in-module
+from sqlalchemy.event import (  # pylint: disable=no-name-in-module
+    listen,
+    remove,
+)
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.sqlalchemy.package import (
@@ -111,6 +114,11 @@ def __init__(
         listen(engine, "after_cursor_execute", _after_cur_exec)
         listen(engine, "handle_error", _handle_error)
 
+    def remove_event_listeners(self):
+        remove(self.engine, "before_cursor_execute", self._before_cur_exec)
+        remove(self.engine, "after_cursor_execute", _after_cur_exec)
+        remove(self.engine, "handle_error", _handle_error)
+
     def _operation_name(self, db_name, statement):
         parts = []
         if isinstance(statement, str):

instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlalchemy.py

@@ -248,6 +248,7 @@ def test_uninstrument(self):
 
         self.memory_exporter.clear()
         SQLAlchemyInstrumentor().uninstrument()
+        cnx.execute("SELECT	1 + 1;").fetchall()
         engine2 = create_engine("sqlite:///:memory:")
         cnx2 = engine2.connect()
         cnx2.execute("SELECT	2 + 2;").fetchall()