Spaces:
Running
Running
File size: 5,230 Bytes
0491e54 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
"""
Linear Client for FocusFlow.
Handles integration with Linear API (or MCP server) for task synchronization.
Falls back to mock data if no API key is provided.
"""
import os
import json
import requests
from typing import List, Dict, Optional
from datetime import datetime
class LinearClient:
"""Client for interacting with Linear."""
def __init__(self, api_key: Optional[str] = None):
"""Initialize Linear client."""
self.api_key = api_key or os.getenv("LINEAR_API_KEY")
self.api_url = "https://api.linear.app/graphql"
self.is_active = bool(self.api_key)
if not self.is_active:
print("ℹ️ Linear: No API key found. Using mock data.")
def _headers(self) -> Dict[str, str]:
"""Get request headers."""
return {
"Content-Type": "application/json",
"Authorization": self.api_key
}
def _query(self, query: str, variables: Dict = None) -> Dict:
"""Execute GraphQL query."""
if not self.is_active:
return {}
try:
response = requests.post(
self.api_url,
headers=self._headers(),
json={"query": query, "variables": variables or {}}
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"⚠️ Linear API error: {e}")
return {}
def get_user_projects(self) -> List[Dict]:
"""Get projects for the current user."""
if not self.is_active:
return [
{"id": "mock-1", "name": "Website Redesign", "description": "Overhaul the company website"},
{"id": "mock-2", "name": "Mobile App", "description": "iOS and Android app development"},
{"id": "mock-3", "name": "API Migration", "description": "Migrate legacy API to GraphQL"}
]
query = """
query {
viewer {
projects(first: 10) {
nodes {
id
name
description
}
}
}
}
"""
result = self._query(query)
try:
return result.get("data", {}).get("viewer", {}).get("projects", {}).get("nodes", [])
except Exception:
return []
def get_project_tasks(self, project_id: str) -> List[Dict]:
"""Get tasks for a specific project."""
if not self.is_active:
# Return mock tasks based on project ID
if project_id == "mock-1":
return [
{"id": "L-101", "title": "Design Homepage", "description": "Create Figma mockups", "estimate": 60},
{"id": "L-102", "title": "Implement Header", "description": "React component for header", "estimate": 30},
{"id": "L-103", "title": "Fix CSS Bugs", "description": "Fix mobile layout issues", "estimate": 45}
]
return [
{"id": "L-201", "title": "Setup Repo", "description": "Initialize git repository", "estimate": 15},
{"id": "L-202", "title": "Basic Auth", "description": "Implement login flow", "estimate": 60}
]
query = """
query($projectId: ID!) {
project(id: $projectId) {
issues(first: 20, filter: { state: { name: { neq: "Done" } } }) {
nodes {
id
title
description
estimate
}
}
}
}
"""
result = self._query(query, {"projectId": project_id})
try:
return result.get("data", {}).get("project", {}).get("issues", {}).get("nodes", [])
except Exception:
return []
def create_task(self, title: str, description: str = "", team_id: str = None) -> Optional[str]:
"""Create a new task (issue) in Linear."""
if not self.is_active:
print(f"ℹ️ Linear (Mock): Created task '{title}'")
return "mock-new-id"
# Note: This requires a team_id. For simplicity, we might need to fetch a default team first.
# This is a simplified implementation.
if not team_id:
# Try to get the first team
team_query = """query { viewer { teams(first: 1) { nodes { id } } } }"""
team_res = self._query(team_query)
try:
team_id = team_res["data"]["viewer"]["teams"]["nodes"][0]["id"]
except:
return None
mutation = """
mutation($title: String!, $description: String, $teamId: String!) {
issueCreate(input: { title: $title, description: $description, teamId: $teamId }) {
issue {
id
}
}
}
"""
result = self._query(mutation, {"title": title, "description": description, "teamId": team_id})
try:
return result["data"]["issueCreate"]["issue"]["id"]
except:
return None
|