Skip to content
  • 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 { ... } 
          }
        }
      }
    }
  • {
      "fullPath": "my-group",
      "searchPrefix": "my-prefix",
      "projCursor": null,
      "groupCursor": null
    }
  • 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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment