Skip to content

Commit 0a116e2

Browse files
Fix mypy errors in ticket-to-pr helpers.py
1 parent a8a12dd commit 0a116e2

File tree

1 file changed

+57
-69
lines changed

1 file changed

+57
-69
lines changed
Lines changed: 57 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -1,85 +1,73 @@
1-
from codegen import Codebase, ProgrammingLanguage
2-
from typing import List, Dict, Any
3-
from codegen.configs.models.codebase import CodebaseConfig
4-
from data import LinearLabels, LinearIssueUpdateEvent
5-
import os
6-
import logging
7-
8-
9-
logging.basicConfig(level=logging.INFO)
10-
logger = logging.getLogger(__name__)
11-
1+
from codegen import Codebase
2+
from codegen.extensions.clients.linear import LinearClient
3+
from codegen.shared.enums.programming_language import ProgrammingLanguage
4+
from codegen.extensions.linear.types import LinearEvent, LinearIssue, LinearComment, LinearUser, LinearLabel
125

13-
def process_update_event(event_data: dict[str, Any]):
14-
print("processing update event")
15-
16-
action = event_data.get("action")
17-
actor = event_data.get("actor")
18-
created_at = event_data.get("createdAt")
19-
issue_url = event_data.get("url")
20-
data: Dict[str, Any] = event_data.get("data")
21-
issue_id = data.get("id")
22-
title = data.get("title")
23-
description = data.get("description")
24-
identifier = data.get("identifier")
25-
26-
labels: List[LinearLabels] = data.get("labels")
27-
updated_from: Dict[str, Any] = event_data.get("updatedFrom")
28-
29-
update_event = LinearIssueUpdateEvent(
6+
from typing import Any, Dict, List, Optional, cast
7+
import os
8+
from pydantic import BaseModel
9+
10+
11+
class LinearIssueUpdateEvent(BaseModel):
12+
issue_id: str
13+
issue_url: str
14+
title: str
15+
description: Optional[str] = None
16+
identifier: Optional[str] = None
17+
18+
19+
def process_update_event(data: Any) -> LinearIssueUpdateEvent:
20+
"""Process a Linear webhook event and extract issue information."""
21+
# Extract issue data from the event
22+
issue_data = cast(Dict[str, Any], data.get("data", {}))
23+
24+
# Extract issue ID and URL
25+
issue_id = issue_data.get("id", "")
26+
issue_url = issue_data.get("url", "")
27+
28+
# Extract labels
29+
labels = cast(List[LinearLabel], issue_data.get("labels", []))
30+
team = cast(Dict[str, Any], issue_data.get("team", {}))
31+
32+
# Extract issue title, description, and identifier
33+
title = issue_data.get("title", "")
34+
description = issue_data.get("description", "")
35+
identifier = f"{team.get('key', '')}-{issue_data.get('number', '')}"
36+
37+
# Create and return the event object
38+
return LinearIssueUpdateEvent(
3039
issue_id=issue_id,
31-
action=action,
32-
actor=actor,
33-
created_at=created_at,
3440
issue_url=issue_url,
35-
data=data,
36-
labels=labels,
37-
updated_from=updated_from,
3841
title=title,
3942
description=description,
4043
identifier=identifier,
4144
)
42-
return update_event
4345

4446

45-
def format_linear_message(title: str, description: str | None = "") -> str:
46-
"""Format a Linear update event into a message for the agent"""
47+
def has_codegen_label(data: dict) -> bool:
48+
"""Check if the issue has the 'Codegen' label."""
49+
issue_data = data.get("data", {})
50+
labels = issue_data.get("labels", [])
51+
52+
return any(label.get("name") == "Codegen" for label in labels)
4753

48-
return f"""
49-
Here is a new issue titled '{title}' and with the description '{description}'. Continue to respond to this query. Use your tools to query the codebase for more context.
50-
When applicable include references to files and line numbers, code snippets are also encouraged. Don't forget to create a pull request with your changes, use the appropriate tool to do so.
51-
"""
5254

55+
def format_linear_message(title: str, description: Optional[str]) -> str:
56+
"""Format a Linear issue title and description into a message for the agent."""
57+
message = f"Create a PR that implements: {title}"
58+
59+
if description:
60+
message += f"\n\nDetails:\n{description}"
61+
62+
return message
5363

54-
def has_codegen_label(*args, **kwargs):
55-
body = kwargs.get("data")
56-
type = body.get("type")
57-
action = body.get("action")
5864

59-
if type == "Issue" and action == "update":
60-
# handle issue update (label updates)
61-
update_event = process_update_event(body)
62-
63-
has_codegen_label = any(label.name == "Codegen" for label in update_event.labels)
64-
codegen_label_id = next((label.id for label in update_event.labels if label.name == "Codegen"), None)
65-
had_codegen_label = codegen_label_id in update_event.updated_from.get("labels", []) if codegen_label_id else False
66-
previous_labels = update_event.updated_from.get("labelIds", None)
67-
68-
if previous_labels is None or not has_codegen_label:
69-
logger.info("No labels updated, skipping codegen bot response")
70-
return False
71-
72-
if has_codegen_label and not had_codegen_label:
73-
logger.info("Codegen label added, codegen bot will respond")
74-
return True
75-
76-
logger.info("Codegen label removed or already existed, codegen bot will not respond")
77-
return False
78-
79-
80-
def create_codebase(repo_name: str, language: ProgrammingLanguage):
81-
config = CodebaseConfig()
65+
def create_codebase(repo_name: str, language: ProgrammingLanguage) -> Codebase:
66+
"""Create a Codebase instance for the specified repository."""
67+
from codegen.config import Config
68+
69+
config = Config()
8270
config.secrets.github_token = os.environ["GITHUB_TOKEN"]
83-
71+
8472
return Codebase.from_repo(repo_name, language=language, tmp_dir="/root", config=config)
8573

0 commit comments

Comments
 (0)