(
self,
query: str,
variables: dict | None = None,
use_sub_issues: bool = False,
for_projects: bool = False,
)
| 73 | return config.github_token if config.github_token else None |
| 74 | |
| 75 | async def _execute( |
| 76 | self, |
| 77 | query: str, |
| 78 | variables: dict | None = None, |
| 79 | use_sub_issues: bool = False, |
| 80 | for_projects: bool = False, |
| 81 | ) -> dict: |
| 82 | token = await self._get_token(for_projects=for_projects) |
| 83 | if not token: |
| 84 | if for_projects: |
| 85 | return { |
| 86 | "error": "ProjectV2 access requires GitHub App with project permissions OR GITHUB_PROJECT_PAT with 'project' scope." |
| 87 | } |
| 88 | return {"error": "GitHub token not configured"} |
| 89 | |
| 90 | headers = { |
| 91 | "Authorization": f"Bearer {token}", |
| 92 | "Content-Type": "application/json", |
| 93 | } |
| 94 | |
| 95 | if use_sub_issues or "subIssue" in query or "parent" in query: |
| 96 | headers["GraphQL-Features"] = "sub_issues" |
| 97 | |
| 98 | payload: dict[str, Any] = {"query": query} |
| 99 | if variables: |
| 100 | payload["variables"] = variables |
| 101 | |
| 102 | try: |
| 103 | session = await self.get_session() |
| 104 | async with session.post( |
| 105 | GITHUB_GRAPHQL_URL, |
| 106 | json=payload, |
| 107 | headers=headers, |
| 108 | timeout=aiohttp.ClientTimeout(total=15), |
| 109 | ) as response: |
| 110 | if response.status == 200: |
| 111 | data = await response.json() |
| 112 | if "errors" in data: |
| 113 | error_msgs = [e.get("message", str(e)) for e in data["errors"]] |
| 114 | logger.warning(f"GraphQL errors: {error_msgs}") |
| 115 | return { |
| 116 | "data": data.get("data"), |
| 117 | "error": "; ".join(error_msgs), |
| 118 | } |
| 119 | return {"data": data.get("data")} |
| 120 | else: |
| 121 | error_text = await response.text() |
| 122 | logger.error(f"GraphQL error {response.status}: {error_text[:200]}") |
| 123 | return {"error": f"GitHub API error {response.status}: {error_text[:100]}"} |
| 124 | except Exception as e: |
| 125 | logger.error(f"GraphQL request failed: {e}") |
| 126 | return {"error": f"GitHub request failed: {str(e)}"} |
| 127 | |
| 128 | async def get_issue_full(self, issue_number: int, comments_count: int = 5) -> dict | None: |
| 129 | query = """ |
no test coverage detected