Run make lint on repo

Author: Ickerday

.basedpyright/baseline.json

@@ -26,11 +26,7 @@ def setup_logging(app_name: str) -> logging.Logger:
     logger = logging.getLogger(app_name)
     logger.setLevel(logging.DEBUG)
 
-    handler = logging.handlers.RotatingFileHandler(
-        LOG_PATH, maxBytes=1024 * 1024, backupCount=5
-    )
-    handler.setFormatter(
-        logging.Formatter(f"%(asctime)s %(levelname)s [{app_name}] %(message)s")
-    )
+    handler = logging.handlers.RotatingFileHandler(LOG_PATH, maxBytes=1024 * 1024, backupCount=5)
+    handler.setFormatter(logging.Formatter(f"%(asctime)s %(levelname)s [{app_name}] %(message)s"))
     logger.addHandler(handler)
     return logger

examples/ai_custom_alert_app/bin/setup_logging.py

@@ -40,9 +40,7 @@
 # one that might not exist on the filesystem. In such case we unset the env, which
 # causes the default Certificate Authorities to be used instead.
 CA_TRUST_STORE = "/opt/splunk/openssl/cert.pem"
-if os.environ.get("SSL_CERT_FILE") == CA_TRUST_STORE and not os.path.exists(
-    CA_TRUST_STORE
-):
+if os.environ.get("SSL_CERT_FILE") == CA_TRUST_STORE and not os.path.exists(CA_TRUST_STORE):
     del os.environ["SSL_CERT_FILE"]
 
 
@@ -86,9 +84,7 @@ class AgenticSeverityAssessment(BaseModel):
     recommended_action: str
 
 
-async def invoke_agent(
-    service: client.Service, alert_data: AlertData
-) -> AgenticSeverityAssessment:
+async def invoke_agent(service: client.Service, alert_data: AlertData) -> AgenticSeverityAssessment:
     async with Agent(
         model=LLM_MODEL,
         system_prompt=SYSTEM_PROMPT,

examples/ai_custom_alert_app/bin/threat_level_assessment.py

@@ -34,7 +34,7 @@
 from splunklib.searchcommands import (
     Configuration,
     Option,
-    dispatch,  # pyright: ignore[reportPrivateLocalImportUsage]
+    dispatch,
     validators,
 )
 from splunklib.searchcommands.eventing_command import EventingCommand
@@ -43,9 +43,7 @@
 # one that might not exist on the filesystem. In such case we unset the env, which
 # causes the default Certificate Authorities to be used instead.
 CA_TRUST_STORE = "/opt/splunk/openssl/cert.pem"
-if os.environ.get("SSL_CERT_FILE") == CA_TRUST_STORE and not os.path.exists(
-    CA_TRUST_STORE
-):
+if os.environ.get("SSL_CERT_FILE") == CA_TRUST_STORE and not os.path.exists(CA_TRUST_STORE):
     del os.environ["SSL_CERT_FILE"]
 
 APP_NAME = "ai_custom_search_app"

examples/ai_custom_search_app/bin/agentic_reporting_csc.py

@@ -27,12 +27,8 @@ def setup_logging(app_name: str) -> logging.Logger:
     logger = logging.getLogger(app_name)
     logger.setLevel(logging.DEBUG)
 
-    handler = logging.handlers.RotatingFileHandler(
-        LOG_FILE, maxBytes=1024 * 1024, backupCount=5
-    )
-    handler.setFormatter(
-        logging.Formatter(f"%(asctime)s %(levelname)s [{app_name}] %(message)s")
-    )
+    handler = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes=1024 * 1024, backupCount=5)
+    handler.setFormatter(logging.Formatter(f"%(asctime)s %(levelname)s [{app_name}] %(message)s"))
     logger.addHandler(handler)
 
     return logger

examples/ai_custom_search_app/bin/setup_logging.py

@@ -44,9 +44,7 @@
 # one that might not exist on the filesystem. In such case we unset the env, which
 # causes the default Certificate Authorities to be used instead.
 CA_TRUST_STORE = "/opt/splunk/openssl/cert.pem"
-if os.environ.get("SSL_CERT_FILE") == CA_TRUST_STORE and not os.path.exists(
-    CA_TRUST_STORE
-):
+if os.environ.get("SSL_CERT_FILE") == CA_TRUST_STORE and not os.path.exists(CA_TRUST_STORE):
     del os.environ["SSL_CERT_FILE"]
 
 

examples/ai_modinput_app/bin/agentic_weather.py

@@ -26,12 +26,8 @@ def setup_logging(app_name: str) -> logging.Logger:
     logger = logging.getLogger(app_name)
     logger.setLevel(logging.DEBUG)
 
-    handler = logging.handlers.RotatingFileHandler(
-        LOG_FILE, maxBytes=1024 * 1024, backupCount=5
-    )
-    handler.setFormatter(
-        logging.Formatter(f"%(asctime)s %(levelname)s [{app_name}] %(message)s")
-    )
+    handler = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes=1024 * 1024, backupCount=5)
+    handler.setFormatter(logging.Formatter(f"%(asctime)s %(levelname)s [{app_name}] %(message)s"))
     logger.addHandler(handler)
 
     return logger

examples/ai_modinput_app/bin/setup_logging.py

@@ -26,9 +26,7 @@
 # To set the logging level of splunklib
 # ex. To enable debug logs, call this method with parameter 'logging.DEBUG'
 # default logging level is set to 'WARNING'
-def setup_logging(
-    level, log_format=DEFAULT_LOG_FORMAT, date_format=DEFAULT_DATE_FORMAT
-):
+def setup_logging(level, log_format=DEFAULT_LOG_FORMAT, date_format=DEFAULT_DATE_FORMAT):
     logging.basicConfig(level=level, format=log_format, datefmt=date_format)
 
 

splunklib/__init__.py

@@ -183,9 +183,7 @@ async def _start_agent(self) -> AsyncGenerator[Self]:
                 "internal error: _impl was not set to None after agent invocation"
             )
 
-            splunk_username = await asyncio.to_thread(
-                lambda: _get_splunk_username(self._service)
-            )
+            splunk_username = await asyncio.to_thread(lambda: _get_splunk_username(self._service))
             _validate_agent_privileges(splunk_username)
 
             self.logger.debug(f"Creating agent {self.name=}; {self.trace_id=}")
@@ -201,9 +199,7 @@ async def _start_agent(self) -> AsyncGenerator[Self]:
 
             self._impl = None
 
-    async def _load_tools(
-        self, stack: AsyncExitStack, splunk_username: str
-    ) -> list[Tool]:
+    async def _load_tools(self, stack: AsyncExitStack, splunk_username: str) -> list[Tool]:
         tools: list[Tool] = []
         if not self.tool_settings.local and not self.tool_settings.remote:
             return tools
@@ -234,9 +230,7 @@ async def _load_tools(
         if self.tool_settings.remote:
             self.logger.debug("Probing MCP Server App availability")
             remote_session = await stack.enter_async_context(
-                connect_remote_mcp(
-                    self._service, app_id, self.trace_id, splunk_username
-                )
+                connect_remote_mcp(self._service, app_id, self.trace_id, splunk_username)
             )
 
             if remote_session:
@@ -252,9 +246,7 @@ async def _load_tools(
                 allowlist = self.tool_settings.remote.allowlist
                 remote_tools = [rt for rt in remote_tools if allowlist.is_allowed(rt)]
 
-                self.logger.debug(
-                    f"Loaded remote_tools={[t.name for t in remote_tools]}"
-                )
+                self.logger.debug(f"Loaded remote_tools={[t.name for t in remote_tools]}")
                 tools.extend(remote_tools)
 
         return tools
@@ -265,13 +257,9 @@ async def __aenter__(self) -> Self:
         self._agent_context_manager = self._start_agent()
         return await self._agent_context_manager.__aenter__()
 
-    async def __aexit__(
-        self, exc_type: ..., exc_value: ..., traceback: ...
-    ) -> bool | None:
+    async def __aexit__(self, exc_type: ..., exc_value: ..., traceback: ...) -> bool | None:
         assert self._agent_context_manager is not None
-        result = await self._agent_context_manager.__aexit__(
-            exc_type, exc_value, traceback
-        )
+        result = await self._agent_context_manager.__aexit__(exc_type, exc_value, traceback)
         self._agent_context_manager = None
         return result
 
@@ -324,9 +312,7 @@ def _local_tools_path() -> tuple[str | None, str]:
         app_id, app_dir = locate_app()
         local_tools_path = build_local_tools_path(app_dir)
 
-    assert app_id is not None, (
-        "_load_tools_from_mcp was mocked, but _testing_app_id not"
-    )
+    assert app_id is not None, "_load_tools_from_mcp was mocked, but _testing_app_id not"
 
     if not os.path.exists(local_tools_path):
         local_tools_path = None

splunklib/ai/agent.py

@@ -21,9 +21,7 @@
 class ConversationStore(Protocol):
     async def get_messages(self, thread_id: str) -> Sequence[BaseMessage]: ...
 
-    async def store_messages(
-        self, thread_id: str, messages: list[BaseMessage]
-    ) -> None: ...
+    async def store_messages(self, thread_id: str, messages: list[BaseMessage]) -> None: ...
 
 
 class InMemoryStore(ConversationStore):