add linear webhook example

codegen-examples/examples/linear_webhooks/.env.template

@@ -0,0 +1,3 @@
+LINEAR_ACCESS_TOKEN="..."
+LINEAR_SIGNING_SECRET="..."
+LINEAR_TEAM_ID="..."

codegen-examples/examples/linear_webhooks/pyproject.toml

@@ -0,0 +1,6 @@
+[project]
+name = "linear-webhook-example"
+version = "0.1.0"
+description = "Linear webhook example for Codegen"
+requires-python = ">=3.13"
+dependencies = ["codegen>=0.22.2", "modal>=0.73.25"]

codegen-examples/examples/linear_webhooks/webhooks.py

@@ -0,0 +1,28 @@
+import modal.running_app
+from codegen.extensions.events.app import CodegenApp
+import modal
+
+image = modal.Image.debian_slim(python_version="3.13").apt_install("git").pip_install("fastapi[standard]", "codegen>=v0.22.2")
+app = CodegenApp(name="test-linear", modal_api_key="", image=image)
+
+# Here is an example implementation of setting up an endpoint for receiving webhook events from Linear.
+# The @app.linear.event() decorator takes care of subscribing to the webhook and also unsubscribing when the deployment spun
+# Load environment variables from .env file
+
+
+@app.cls(secrets=[modal.Secret.from_dotenv()], keep_warm=1)
+class LinearEventHandlers:
+    @modal.enter()
+    def enter(self):
+        app.linear.subscribe_all_handlers()
+
+    @modal.exit()
+    def exit(self):
+        app.linear.unsubscribe_all_handlers()
+
+    @modal.web_endpoint(method="POST")
+    @app.linear.event("Issue")
+    def test(self, data: dict):
+        # handle webhook event
+        # data is the payload of the webhook event
+        print(data)

src/codegen/extensions/clients/linear.py

@@ -130,7 +130,7 @@ def unregister_webhook(self, webhook_id: str):
         response = requests.post(self.api_endpoint, headers=self.api_headers, json={"query": mutation, "variables": variables})
         return response.json()
 
-    def register_webhook(self, webhook_url: str, team_id: str, secret: str, enabled: bool, resource_types: list[str]):
+    def register_webhook(self, webhook_url: str, team_id: str, secret: str, enabled: bool, resource_types: list[str]) -> str | None:
         mutation = """
             mutation createWebhook($input: WebhookCreateInput!) {
                 webhookCreate(input: $input) {

src/codegen/extensions/events/linear.py

@@ -15,6 +15,7 @@
 class RegisteredWebhookHandler(BaseModel):
     webhook_id: str | None = None
     handler_func: Callable
+    event_name: str
 
 
 class Linear(EventHandlerManagerProtocol):
@@ -23,15 +24,22 @@
         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 = {}
+        self._webhook_url = None
 
-    def subscribe_handler_to_webhook(self, web_url: str, event_name: str):
+    def subscribe_handler_to_webhook(self, handler: RegisteredWebhookHandler):
         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)
+        web_url = modal.Function.from_name(app_name=self.app.name, name=handler.handler_func.__qualname__).web_url
+        result = client.register_webhook(team_id=self.linear_team_id, webhook_url=web_url, enabled=True, resource_types=[handler.event_name], secret=self.signing_secret)
         return result
 
+    def subscribe_all_handlers(self):
+        for handler_key in self.registered_handlers:
+            handler = self.registered_handlers[handler_key]
+            result = self.subscribe_handler_to_webhook(handler=self.registered_handlers[handler_key])
+            handler.webhook_id = result
+
     def unsubscribe_handler_to_webhook(self, registered_handler: RegisteredWebhookHandler):
         webhook_id = registered_handler.webhook_id
 
         client = LinearClient(access_token=self.access_token)
@@ -59,13 +67,7 @@
             # Register the handler with the app's registry.
             modal_ready_func = func
             func_name = func.__qualname__
-            app_name = self.app.name
-            web_url = modal.Function.from_name(app_name=app_name, name=func_name).web_url
-
-            self.registered_handlers[func_name] = RegisteredWebhookHandler(handler_func=modal_ready_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
+            self.registered_handlers[func_name] = RegisteredWebhookHandler(handler_func=modal_ready_func, event_name=event_name)
 
             @functools.wraps(func)
             def wrapper(*args, **kwargs):