Automated pre-commit update

Author: rushilpatel0

src/codegen/extensions/clients/linear.py

@@ -1,23 +1,26 @@
 import json
 import logging
-import requests
 
+import requests
 from pydantic import BaseModel
 
 logger = logging.getLogger(__name__)
 
 
 # --- TYPES
 
+
 class LinearUser(BaseModel):
     id: str
     name: str
 
+
 class LinearComment(BaseModel):
     id: str
     body: str
     user: LinearUser | None = None
 
+
 class LinearIssue(BaseModel):
     id: str
     title: str
@@ -49,12 +52,7 @@ def get_issue(self, issue_id: str) -> LinearIssue:
         response = requests.post(self.api_endpoint, headers=self.api_headers, json={"query": query, "variables": variables})
         data = response.json()
         issue_data = data["data"]["issue"]
-        return LinearIssue(
-            id=issue_data["id"],
-            title=issue_data["title"],
-            description=issue_data["description"]
-        )
-        
+        return LinearIssue(id=issue_data["id"], title=issue_data["title"], description=issue_data["description"])
 
     def get_issue_comments(self, issue_id: str) -> list[LinearComment]:
         query = """
@@ -69,7 +67,7 @@ def get_issue_comments(self, issue_id: str) -> list[LinearComment]:
                             name
                         }
                     }
-                        
+
                     }
                 }
             }
@@ -83,21 +81,13 @@ def get_issue_comments(self, issue_id: str) -> list[LinearComment]:
         parsed_comments = []
         for comment in comments:
             user = comment.get("user", None)
-            parsed_comment = LinearComment(
-                id=comment["id"],
-                body=comment["body"],
-                user=LinearUser(
-                    id=user.get("id"),
-                    name=user.get("name")
-                ) if user else None
-            )
+            parsed_comment = LinearComment(id=comment["id"], body=comment["body"], user=LinearUser(id=user.get("id"), name=user.get("name")) if user else None)
             parsed_comments.append(parsed_comment)
 
         # Convert raw comments to LinearComment objects
         return parsed_comments
 
-
-    def comment_on_issue(self, issue_id: str, body: str) -> dict:   
+    def comment_on_issue(self, issue_id: str, body: str) -> dict:
         """issue_id is our internal issue ID"""
         query = """mutation makeComment($issueId: String!, $body: String!) {
           commentCreate(input: {issueId: $issueId, body: $body}) {
@@ -106,7 +96,7 @@ def comment_on_issue(self, issue_id: str, body: str) -> dict:
               body
               url
               user {
-                id  
+                id
                 name
               }
             }
@@ -125,8 +115,8 @@ def comment_on_issue(self, issue_id: str, body: str) -> dict:
 
             return comment_data
         except:
-            raise Exception(f"Error creating comment\n{data}")
-
+            msg = f"Error creating comment\n{data}"
+            raise Exception(msg)
 
     def unregister_webhook(self, webhook_id: str):
         mutation = """
@@ -153,18 +143,20 @@ def register_webhook(self, webhook_url: str, team_id: str, secret: str, enabled:
             }
         """
 
-        variables = {"input": {
-            "url": webhook_url,
-            "teamId": team_id,
-            "resourceTypes": resource_types,
-            "enabled": enabled,
-            "secret": secret,
-        }}
-        
+        variables = {
+            "input": {
+                "url": webhook_url,
+                "teamId": team_id,
+                "resourceTypes": resource_types,
+                "enabled": enabled,
+                "secret": secret,
+            }
+        }
+
         response = requests.post(self.api_endpoint, headers=self.api_headers, json={"query": mutation, "variables": variables})
         if response.status_code != 200:
             return None
-        
+
         body = response.json()
-        body = body["data"]["webhookCreate"]["webhook"]['id']
+        body = body["data"]["webhookCreate"]["webhook"]["id"]
         return body

src/codegen/extensions/events/app.py

@@ -1,13 +1,13 @@
-import modal
 import logging
-from codegen.extensions.events.linear import Linear
 
+import modal
+
+from codegen.extensions.events.linear import Linear
 
 logger = logging.getLogger(__name__)
 
 
 class CodegenApp(modal.App):
-
     linear: Linear
 
     def __init__(self, name, modal_api_key, image: modal.Image):
@@ -19,5 +19,3 @@ def __init__(self, name, modal_api_key, image: modal.Image):
 
         # Expose a attribute that provides the event decorator for different providers.
         self.linear = Linear(self)
-
-

src/codegen/extensions/events/interface.py

@@ -1,10 +1,12 @@
-
 from typing import Protocol
+
 import modal
 
+
 class EventHandlerManagerProtocol(Protocol):
     def subscribe_handler_to_webhook(self, func_name: str, modal_app: modal.App, event_name):
         pass
+
     def unsubscribe_handler_to_webhook(self, func_name: str, modal_app: modal.App, event_name):
         pass
 

src/codegen/extensions/events/linear.py

@@ -1,13 +1,13 @@
-from typing import TYPE_CHECKING, Callable
-from anthropic import BaseModel
 import functools
+import logging
 import os
+from typing import Callable
+
 import modal
-import logging
+from anthropic import BaseModel
 
-from codegen.extensions.events.interface import EventHandlerManagerProtocol
 from codegen.extensions.clients.linear import LinearClient
-
+from codegen.extensions.events.interface import EventHandlerManagerProtocol
 
 logger = logging.getLogger(__name__)
 
@@ -16,28 +16,24 @@ class RegisteredWebhookHandler(BaseModel):
     webhook_id: str | None = None
     handler_func: Callable
 
+
 class Linear(EventHandlerManagerProtocol):
     def __init__(self, app: modal.App):
         self.app = app
-        self.access_token= os.environ["LINEAR_ACCESS_TOKEN"] # move to extensions config.
+        self.access_token = os.environ["LINEAR_ACCESS_TOKEN"]  # move to extensions config.
         self.signing_secret = os.environ["LINEAR_SIGNING_SECRET"]
         self.linear_team_id = os.environ["LINEAR_TEAM_ID"]
         self.registered_handlers = {}
 
     def subscribe_handler_to_webhook(self, web_url: str, event_name: str):
         client = LinearClient(access_token=self.access_token)
 
-        result = client.register_webhook(
-            team_id=self.linear_team_id, 
-            webhook_url=web_url, 
-            enabled=True, 
-            resource_types=[event_name], 
-            secret=self.signing_secret)
+        result = client.register_webhook(team_id=self.linear_team_id, webhook_url=web_url, enabled=True, resource_types=[event_name], secret=self.signing_secret)
         return result
-        
+
     def unsubscribe_handler_to_webhook(self, registered_handler: RegisteredWebhookHandler):
         webhook_id = registered_handler.webhook_id
-       
+
         client = LinearClient(access_token=self.access_token)
         if webhook_id:
             print(f"Unsubscribing from webhook {webhook_id}")
@@ -46,20 +42,19 @@ def unsubscribe_handler_to_webhook(self, registered_handler: RegisteredWebhookHa
         else:
             print("No webhook id found for handler")
             return None
+
     def unsubscribe_all_handlers(self):
         for handler in self.registered_handlers:
             self.unsubscribe_handler_to_webhook(self.registered_handlers[handler])
 
-
-
     def event(self, event_name):
-        """
-        Decorator for registering an event handler.
+        """Decorator for registering an event handler.
 
         :param event_name: The name of the event to handle.
         :param register_hook: An optional function to call during registration,
                               e.g., to make an API call to register the webhook.
         """
+
         def decorator(func):
             # Register the handler with the app's registry.
             modal_ready_func = func
@@ -71,9 +66,11 @@ def decorator(func):
 
             webhook_id = self.subscribe_handler_to_webhook(web_url=web_url, event_name=event_name)
             self.registered_handlers[func_name].webhook_id = webhook_id
+
             @functools.wraps(func)
             def wrapper(*args, **kwargs):
                 return func(*args, **kwargs)
+
             return wrapper
-                    
+
         return decorator