fix: clean up whitespace and fix implementation

Author: RiyaChaturvedi37

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

@@ -51,10 +51,10 @@ async def main():
 
 import re
 from typing import Collection
- 
+
 import asyncpg
 import wrapt
- 
+
 from opentelemetry import trace
 from opentelemetry.instrumentation.asyncpg.package import _instruments
 from opentelemetry.instrumentation.asyncpg.version import __version__
@@ -75,23 +75,23 @@ async def main():
 )
 from opentelemetry.trace import SpanKind
 from opentelemetry.trace.status import Status, StatusCode
- 
- 
+
+
 def _hydrate_span_from_args(connection, query, parameters) -> dict:
     """Get network and database attributes from connection."""
     span_attributes = {DB_SYSTEM: DbSystemValues.POSTGRESQL.value}
- 
+
     # connection contains _params attribute which is a namedtuple ConnectionParameters.
     # https://github.com/MagicStack/asyncpg/blob/master/asyncpg/connection.py#L68
- 
+
     params = getattr(connection, "_params", None)
     dbname = getattr(params, "database", None)
     if dbname:
         span_attributes[DB_NAME] = dbname
     user = getattr(params, "user", None)
     if user:
         span_attributes[DB_USER] = user
- 
+
     # connection contains _addr attribute which is either a host/port tuple, or unix socket string
     # https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html
     addr = getattr(connection, "_addr", None)
@@ -102,16 +102,16 @@ def _hydrate_span_from_args(connection, query, parameters) -> dict:
     elif isinstance(addr, str):
         span_attributes[NET_PEER_NAME] = addr
         span_attributes[NET_TRANSPORT] = NetTransportValues.OTHER.value
- 
+
     if query is not None:
         span_attributes[DB_STATEMENT] = query
- 
+
     if parameters is not None and len(parameters) > 0:
         span_attributes["db.statement.parameters"] = str(parameters)
- 
+
     return span_attributes
- 
- 
+
+
 class AsyncPGInstrumentor(BaseInstrumentor):
     _leading_comment_remover = re.compile(r"^/\*.*?\*/")
     _CLEANUP_QUERIES = frozenset([
@@ -121,20 +121,20 @@ class AsyncPGInstrumentor(BaseInstrumentor):
         "RESET ALL",
     ])
     _tracer = None
- 
+
     def _is_cleanup_query(self, query: str) -> bool:
         if query is None:
             return False
         return any(q in query for q in self._CLEANUP_QUERIES)
- 
+
     def __init__(self, capture_parameters=False, capture_connection_cleanup=True):
         super().__init__()
         self.capture_parameters = capture_parameters
         self.capture_connection_cleanup = capture_connection_cleanup
- 
+
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
- 
+
     def _instrument(self, **kwargs):
         tracer_provider = kwargs.get("tracer_provider")
         self._tracer = trace.get_tracer(
@@ -143,7 +143,7 @@ def _instrument(self, **kwargs):
             tracer_provider,
             schema_url="https://opentelemetry.io/schemas/1.11.0",
         )
- 
+
         for method in [
             "Connection.execute",
             "Connection.executemany",
@@ -154,7 +154,7 @@ def _instrument(self, **kwargs):
             wrapt.wrap_function_wrapper(
                 "asyncpg.connection", method, self._do_execute
             )
- 
+
         for method in [
             "Cursor.fetch",
             "Cursor.forward",
@@ -164,7 +164,7 @@ def _instrument(self, **kwargs):
             wrapt.wrap_function_wrapper(
                 "asyncpg.cursor", method, self._do_cursor_execute
             )
- 
+
     def _uninstrument(self, **__):
         for cls, methods in [
             (
@@ -176,30 +176,29 @@ def _uninstrument(self, **__):
         ]:
             for method_name in methods:
                 unwrap(cls, method_name)
- 
+
     async def _do_execute(self, func, instance, args, kwargs):
         exception = None
         params = getattr(instance, "_params", None)
         name = (
             args[0] if args[0] else getattr(params, "database", "postgresql")
         )
- 
+
         try:
             # Strip leading comments so we get the operation name.
             name = self._leading_comment_remover.sub("", name).split()[0]
         except IndexError:
             name = ""
- 
+
         # Hydrate attributes before span creation to enable filtering
         span_attributes = _hydrate_span_from_args(
             instance,
             args[0],
             args[1:] if self.capture_parameters else None,
         )
- 
         if not self.capture_connection_cleanup and self._is_cleanup_query(args[0]):
             return await func(*args, **kwargs)
- 
+        
         with self._tracer.start_as_current_span(
             name, kind=SpanKind.CLIENT, attributes=span_attributes
         ) as span:
@@ -211,9 +210,9 @@ async def _do_execute(self, func, instance, args, kwargs):
             finally:
                 if span.is_recording() and exception is not None:
                     span.set_status(Status(StatusCode.ERROR))
- 
+
         return result
- 
+
     async def _do_cursor_execute(self, func, instance, args, kwargs):
         """Wrap cursor based functions. For every call this will generate a new span."""
         exception = None
@@ -223,20 +222,20 @@ async def _do_cursor_execute(self, func, instance, args, kwargs):
             if instance._query
             else getattr(params, "database", "postgresql")
         )
- 
+
         try:
             # Strip leading comments so we get the operation name.
             name = self._leading_comment_remover.sub("", name).split()[0]
         except IndexError:
             name = ""
- 
+
         # Hydrate attributes before span creation to enable filtering
         span_attributes = _hydrate_span_from_args(
             instance._connection,
             instance._query,
             instance._args if self.capture_parameters else None,
         )
- 
+
         stop = False
         with self._tracer.start_as_current_span(
             f"CURSOR: {name}",
@@ -254,8 +253,7 @@ async def _do_cursor_execute(self, func, instance, args, kwargs):
             finally:
                 if span.is_recording() and exception is not None:
                     span.set_status(Status(StatusCode.ERROR))
- 
+
         if not stop:
             return result
         raise StopAsyncIteration
-