refactor: optimize logging performance and modernize string formatting

- Replace f-strings with % formatting in logging calls for lazy evaluation
- Convert .format() to f-strings in server/models.py __repr__ methods
- Add test_venv/ to .gitignore
- Improves performance by avoiding string formatting when log levels are disabled

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: ahoblitz

.gitignore

@@ -6,6 +6,7 @@ __pycache__
 .pytest_cache
 .ruff_cache
 .venv
+test_venv/
 coverage.xml
 .nox
 spec.json

src/a2a/server/events/event_consumer.py

@@ -62,7 +62,7 @@ async def consume_one(self) -> Event:
                 InternalError(message='Agent did not return any response')
             ) from e
 
-        logger.debug(f'Dequeued event of type: {type(event)} in consume_one.')
+        logger.debug('Dequeued event of type: %s in consume_one.', type(event))
 
         self.queue.task_done()
 
@@ -95,7 +95,7 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                     self.queue.dequeue_event(), timeout=self._timeout
                 )
                 logger.debug(
-                    f'Dequeued event of type: {type(event)} in consume_all.'
+                    'Dequeued event of type: %s in consume_all.', type(event)
                 )
                 self.queue.task_done()
                 logger.debug(

src/a2a/server/events/event_queue.py

@@ -54,7 +54,7 @@ async def enqueue_event(self, event: Event) -> None:
                 logger.warning('Queue is closed. Event will not be enqueued.')
                 return
 
-        logger.debug(f'Enqueuing event of type: {type(event)}')
+        logger.debug('Enqueuing event of type: %s', type(event))
 
         # Make sure to use put instead of put_nowait to avoid blocking the event loop.
         await self.queue.put(event)
@@ -98,13 +98,13 @@ async def dequeue_event(self, no_wait: bool = False) -> Event:
             logger.debug('Attempting to dequeue event (no_wait=True).')
             event = self.queue.get_nowait()
             logger.debug(
-                f'Dequeued event (no_wait=True) of type: {type(event)}'
+                'Dequeued event (no_wait=True) of type: %s', type(event)
             )
             return event
 
         logger.debug('Attempting to dequeue event (waiting).')
         event = await self.queue.get()
-        logger.debug(f'Dequeued event (waited) of type: {type(event)}')
+        logger.debug('Dequeued event (waited) of type: %s', type(event))
         return event
 
     def task_done(self) -> None:

src/a2a/server/models.py

@@ -147,14 +147,9 @@ def task_metadata(cls) -> Mapped[dict[str, Any] | None]:
     @override
     def __repr__(self) -> str:
         """Return a string representation of the task."""
-        repr_template = (
-            '<{CLS}(id="{ID}", context_id="{CTX_ID}", status="{STATUS}")>'
-        )
-        return repr_template.format(
-            CLS=self.__class__.__name__,
-            ID=self.id,
-            CTX_ID=self.context_id,
-            STATUS=self.status,
+        return (
+            f'<{self.__class__.__name__}(id="{self.id}", '
+            f'context_id="{self.context_id}", status="{self.status}")>'
         )
 
 
@@ -188,12 +183,9 @@ class TaskModel(TaskMixin, base):  # type: ignore
         @override
         def __repr__(self) -> str:
             """Return a string representation of the task."""
-            repr_template = '<TaskModel[{TABLE}](id="{ID}", context_id="{CTX_ID}", status="{STATUS}")>'
-            return repr_template.format(
-                TABLE=table_name,
-                ID=self.id,
-                CTX_ID=self.context_id,
-                STATUS=self.status,
+            return (
+                f'<TaskModel[{table_name}](id="{self.id}", '
+                f'context_id="{self.context_id}", status="{self.status}")>'
             )
 
     # Set a dynamic name for better debugging
@@ -221,11 +213,9 @@ class PushNotificationConfigMixin:
     @override
     def __repr__(self) -> str:
         """Return a string representation of the push notification config."""
-        repr_template = '<{CLS}(task_id="{TID}", config_id="{CID}")>'
-        return repr_template.format(
-            CLS=self.__class__.__name__,
-            TID=self.task_id,
-            CID=self.config_id,
+        return (
+            f'<{self.__class__.__name__}(task_id="{self.task_id}", '
+            f'config_id="{self.config_id}")>'
         )
 
 
@@ -241,11 +231,9 @@ class PushNotificationConfigModel(PushNotificationConfigMixin, base):  # type: i
         @override
         def __repr__(self) -> str:
             """Return a string representation of the push notification config."""
-            repr_template = '<PushNotificationConfigModel[{TABLE}](task_id="{TID}", config_id="{CID}")>'
-            return repr_template.format(
-                TABLE=table_name,
-                TID=self.task_id,
-                CID=self.config_id,
+            return (
+                f'<PushNotificationConfigModel[{table_name}]('
+                f'task_id="{self.task_id}", config_id="{self.config_id}")>'
             )
 
     PushNotificationConfigModel.__name__ = (

src/a2a/server/tasks/database_task_store.py

@@ -124,7 +124,7 @@ async def save(self, task: Task) -> None:
         db_task = self._to_orm(task)
         async with self.async_session_maker.begin() as session:
             await session.merge(db_task)
-            logger.debug(f'Task {task.id} saved/updated successfully.')
+            logger.debug('Task %s saved/updated successfully.', task.id)
 
     async def get(self, task_id: str) -> Task | None:
         """Retrieves a task from the database by ID."""
@@ -135,10 +135,10 @@ async def get(self, task_id: str) -> Task | None:
             task_model = result.scalar_one_or_none()
             if task_model:
                 task = self._from_orm(task_model)
-                logger.debug(f'Task {task_id} retrieved successfully.')
+                logger.debug('Task %s retrieved successfully.', task_id)
                 return task
 
-            logger.debug(f'Task {task_id} not found in store.')
+            logger.debug('Task %s not found in store.', task_id)
             return None
 
     async def delete(self, task_id: str) -> None:
@@ -151,8 +151,8 @@ async def delete(self, task_id: str) -> None:
             # Commit is automatic when using session.begin()
 
             if result.rowcount > 0:
-                logger.info(f'Task {task_id} deleted successfully.')
+                logger.info('Task %s deleted successfully.', task_id)
             else:
                 logger.warning(
-                    f'Attempted to delete nonexistent task with id: {task_id}'
+                    'Attempted to delete nonexistent task with id: %s', task_id
                 )

src/a2a/utils/error_handlers.py

@@ -81,7 +81,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> Response:
                 content={'message': error.message}, status_code=http_code
             )
         except Exception as e:  # noqa: BLE001
-            logger.log(logging.ERROR, f'Unknown error occurred {e}')
+            logger.log(logging.ERROR, 'Unknown error occurred %s', e)
             return JSONResponse(
                 content={'message': 'unknown exception'}, status_code=500
             )

src/a2a/utils/helpers.py

@@ -142,7 +142,7 @@ def decorator(function: Callable) -> Callable:
             async def async_wrapper(self: Any, *args, **kwargs) -> Any:
                 if not expression(self):
                     final_message = error_message or str(expression)
-                    logger.error(f'Unsupported Operation: {final_message}')
+                    logger.error('Unsupported Operation: %s', final_message)
                     raise ServerError(
                         UnsupportedOperationError(message=final_message)
                     )
@@ -154,7 +154,7 @@ async def async_wrapper(self: Any, *args, **kwargs) -> Any:
         def sync_wrapper(self: Any, *args, **kwargs) -> Any:
             if not expression(self):
                 final_message = error_message or str(expression)
-                logger.error(f'Unsupported Operation: {final_message}')
+                logger.error('Unsupported Operation: %s', final_message)
                 raise ServerError(
                     UnsupportedOperationError(message=final_message)
                 )
@@ -186,7 +186,7 @@ def decorator(function):
         async def wrapper(self, *args, **kwargs):
             if not expression(self):
                 final_message = error_message or str(expression)
-                logger.error(f'Unsupported Operation: {final_message}')
+                logger.error('Unsupported Operation: %s', final_message)
                 raise ServerError(
                     UnsupportedOperationError(message=final_message)
                 )

src/a2a/utils/telemetry.py

@@ -171,7 +171,7 @@ def trace_function(  # noqa: PLR0915
     is_async_func = inspect.iscoroutinefunction(func)
 
     logger.debug(
-        f'Start tracing for {actual_span_name}, is_async_func {is_async_func}'
+        'Start tracing for %s, is_async_func %s', actual_span_name, is_async_func
     )
 
     @functools.wraps(func)
@@ -196,7 +196,7 @@ async def async_wrapper(*args, **kwargs) -> Any:
             # asyncio.CancelledError extends from BaseException
             except asyncio.CancelledError as ce:
                 exception = None
-                logger.debug(f'CancelledError in span {actual_span_name}')
+                logger.debug('CancelledError in span %s', actual_span_name)
                 span.record_exception(ce)
                 raise
             except Exception as e:
@@ -212,7 +212,7 @@ async def async_wrapper(*args, **kwargs) -> Any:
                         )
                     except Exception:
                         logger.exception(
-                            f'attribute_extractor error in span {actual_span_name}'
+                            'attribute_extractor error in span %s', actual_span_name
                         )
             return result
 
@@ -246,7 +246,7 @@ def sync_wrapper(*args, **kwargs) -> Any:
                         )
                     except Exception:
                         logger.exception(
-                            f'attribute_extractor error in span {actual_span_name}'
+                            'attribute_extractor error in span %s', actual_span_name
                         )
             return result
 
@@ -307,7 +307,7 @@ def not_traced_method(self):
                 pass
         ```
     """
-    logger.debug(f'Trace all class {include_list}, {exclude_list}')
+    logger.debug('Trace all class %s, %s', include_list, exclude_list)
     exclude_list = exclude_list or []
 
     def decorator(cls: Any) -> Any: