-
query getGroupProjectsRecursive($fullPath: ID!, $searchPrefix: String!, $projCursor: String, $groupCursor: String) { group(fullPath: $fullPath) { id name fullPath # Flat list of projects in this group & all subgroups projects(search: $searchPrefix, includeSubgroups: true, first: 50, after: $projCursor) { pageInfo { hasNextPage endCursor } nodes { id name fullPath description labels(first: 50) { nodes { title } } repository { readme { name path plainData htmlBlob } tags(first: 20) { nodes { name message targetCommit { id shortId title authoredDate authorName } } } } } } # Optional: to traverse subgroup structure descendantGroups(search: $searchPrefix, first: 20, after: $groupCursor) { pageInfo { hasNextPage endCursor } nodes { id name fullPath # You can nest this query again recursively if your GraphQL client supports it projects(search: $searchPrefix, includeSubgroups: true, first: 50) { nodes { id name fullPath description labels(first: 50) { nodes { title } } repository { readme { name path plainData htmlBlob } tags(first: 20) { nodes { name message targetCommit { id shortId title authoredDate authorName } } } } } } # You could go deeper: descendantGroups { ... } } } } }
-
from fastapi import Query @app.get("/gitlab/query") def run_gitlab_query( fullPath: str = Query("default-group", description="GitLab group full path"), searchPrefix: str = Query("", description="Optional project name prefix") ): """ Fetches active GitLab projects (filters out 'deprecated' labels). Uses defaults if query parameters are not provided. """ token = os.getenv("GITLAB_TOKEN") if not token: raise HTTPException(status_code=500, detail="GITLAB_TOKEN environment variable not set") service = GitLabGraphQLService(token) query = service.load_query("query.json") variables = { "fullPath": fullPath, "searchPrefix": searchPrefix, "projCursor": None # pagination starts at beginning } active_projects = service.execute_query(query, variables) return {"data": active_projects}
-
class GitLabGraphQLService: """ A service class for invoking GitLab's GraphQL API with pagination and optional filtering of deprecated projects. """ def __init__(self, token: str): self.endpoint = "https://gitlab.com/api/graphql" self.headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } def load_query(self, query_file: str) -> str: try: file_path = os.path.join(os.path.dirname(__file__), query_file) with open(file_path, "r", encoding="utf-8") as file: data = json.load(file) query = data.get("query") if not query: raise ValueError("The JSON file must contain a 'query' key.") return query except Exception as e: logger.error(f"Error loading query file: {e}") raise HTTPException(status_code=500, detail=str(e)) def execute_query(self, query: str, variables: dict = None, remove_deprecated: bool = True): """ Executes the GraphQL query, handles pagination, and filters out deprecated projects if requested. """ active_projects = [] proj_cursor = variables.get("projCursor") if variables else None while True: payload = { "query": query, "variables": {**(variables or {}), "projCursor": proj_cursor} } try: response = requests.post(self.endpoint, headers=self.headers, json=payload) response.raise_for_status() result = response.json() if "errors" in result: logger.error(f"GraphQL errors: {result['errors']}") raise HTTPException(status_code=400, detail=result["errors"]) group_data = result["data"]["group"] projects_data = group_data["projects"]["nodes"] page_info = group_data["projects"]["pageInfo"] # Filter out deprecated projects for project in projects_data: if remove_deprecated: labels = [label["title"].lower() for label in project["labels"]["nodes"]] if "deprecated" in labels: continue active_projects.append(project) if page_info["hasNextPage"]: proj_cursor = page_info["endCursor"] else: break except requests.exceptions.HTTPError as http_err: logger.error(f"HTTP error: {http_err}") raise HTTPException(status_code=response.status_code, detail=str(http_err)) except requests.exceptions.RequestException as req_err: logger.error(f"Request error: {req_err}") raise HTTPException(status_code=500, detail=str(req_err)) except Exception as e: logger.error(f"Unexpected error: {e}") raise HTTPException(status_code=500, detail=str(e)) return active_projects
Please register or sign in to comment