[Integration][Gitlab-v2] Added support for file kind

integrations/gitlab-v2/CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- towncrier release notes start -->
 
+## 0.1.1-dev (2025-03-12)
+
+### Improvements
+
+- Added Support for file kind
+
+
 ## 0.1.0-dev (2025-02-27)
 
 ### Features

integrations/gitlab-v2/clients/gitlab_client.py

@@ -2,13 +2,16 @@
 from functools import partial
 from typing import Any, AsyncIterator, Callable, Optional
 
+import anyio
 from loguru import logger
 from port_ocean.utils.async_iterators import (
     semaphore_async_iterator,
     stream_async_iterators_tasks,
 )
+from urllib.parse import quote
 
 from .rest_client import RestClient
+from .utils import parse_file_content
 
 
 class GitLabClient:
@@ -21,6 +24,7 @@ class GitLabClient:
     def __init__(self, base_url: str, token: str) -> None:
         self.rest = RestClient(base_url, token, endpoint="api/v4")
 
+    # Public: Project Methods
     async def get_projects(
         self,
         params: Optional[dict[str, Any]] = None,
@@ -37,39 +41,12 @@ async def get_projects(
 
             if include_languages:
                 enriched_batch = await self._enrich_batch(
-                    enriched_batch, self.enrich_project_with_languages, max_concurrent
+                    enriched_batch, self._enrich_project_with_languages, max_concurrent
                 )
 
             yield enriched_batch
 
-    async def _enrich_batch(
-        self,
-        batch: list[dict[str, Any]],
-        enrich_func: Callable[[dict[str, Any]], AsyncIterator[list[dict[str, Any]]]],
-        max_concurrent: int,
-    ) -> list[dict[str, Any]]:
-
-        semaphore = asyncio.Semaphore(max_concurrent)
-        tasks = [
-            semaphore_async_iterator(semaphore, partial(enrich_func, project))
-            for project in batch
-        ]
-        enriched_projects = []
-        async for enriched_batch in stream_async_iterators_tasks(*tasks):
-            enriched_projects.extend(enriched_batch)
-        return enriched_projects
-
-    async def enrich_project_with_languages(
-        self, project: dict[str, Any]
-    ) -> AsyncIterator[list[dict[str, Any]]]:
-
-        project_path = project.get("path_with_namespace", str(project["id"]))
-        logger.debug(f"Enriching {project_path} with languages")
-        languages = await self.rest.get_project_languages(project_path)
-        logger.info(f"Fetched languages for {project_path}: {languages}")
-        project["__languages"] = languages
-        yield [project]
-
+    # Public: Group Methods
     async def get_groups(self) -> AsyncIterator[list[dict[str, Any]]]:
         """Fetch all groups accessible to the user."""
         async for batch in self.rest.get_paginated_resource(
@@ -83,7 +60,6 @@ async def get_groups_resource(
         resource_type: str,
         max_concurrent: int = 10,
     ) -> AsyncIterator[list[dict[str, Any]]]:
-
         semaphore = asyncio.Semaphore(max_concurrent)
         tasks = [
             semaphore_async_iterator(
@@ -96,6 +72,79 @@ async def get_groups_resource(
             if batch:
                 yield batch
 
+    # Public: File Methods
+    async def get_file_content(
+        self, project_id: str, file_path: str, ref: str = "main"
+    ) -> Optional[str]:
+        return await self.rest.get_file_content(project_id, file_path, ref)
+
+    async def file_exists(self, project_id: str, scope: str, query: str) -> bool:
+        params = {"scope": scope, "search": query}
+        encoded_project_path = quote(project_id, safe="")
+        response = await self.rest.send_api_request(
+            "GET", f"projects/{encoded_project_path}/search", params
+        )
+        return bool(response)
+
+    async def search_files(
+        self,
+        scope: str,
+        query: str,
+        repositories: Optional[list[str]] = None,
+        max_concurrent: int = 10,
+    ) -> AsyncIterator[list[dict[str, Any]]]:
+        """Search for files in repositories or groups using a scope and query, returning files with content."""
+        logger.info(f"Searching for files with scope '{scope}' and query '{query}'")
+        semaphore = asyncio.Semaphore(max_concurrent)
+
+        if repositories:
+            logger.info(f"Searching in {len(repositories)} repositories")
+            for repo in repositories:
+                logger.debug(f"Searching repo '{repo}' for query '{query}'")
+                async for batch in self._search_in_repository(repo, scope, query):
+                    yield batch
+        else:
+            logger.info("Searching across all accessible groups")
+            async for groups in self.get_groups():
+                logger.debug(f"Processing batch of {len(groups)} groups")
+                tasks = [
+                    semaphore_async_iterator(
+                        semaphore,
+                        partial(self._search_in_group, group, scope, query),
+                    )
+                    for group in groups
+                ]
+                async for batch in stream_async_iterators_tasks(*tasks):
+                    yield batch
+
+    # Helpers: Enrichment
+    async def _enrich_batch(
+        self,
+        batch: list[dict[str, Any]],
+        enrich_func: Callable[[dict[str, Any]], AsyncIterator[list[dict[str, Any]]]],
+        max_concurrent: int,
+    ) -> list[dict[str, Any]]:
+        semaphore = asyncio.Semaphore(max_concurrent)
+        tasks = [
+            semaphore_async_iterator(semaphore, partial(enrich_func, project))
+            for project in batch
+        ]
+        enriched_projects = []
+        async for enriched_batch in stream_async_iterators_tasks(*tasks):
+            enriched_projects.extend(enriched_batch)
+        return enriched_projects
+
+    async def _enrich_project_with_languages(
+        self, project: dict[str, Any]
+    ) -> AsyncIterator[list[dict[str, Any]]]:
+        project_path = project.get("path_with_namespace", str(project["id"]))
+        logger.debug(f"Enriching {project_path} with languages")
+        languages = await self.rest.get_project_languages(project_path)
+        logger.info(f"Fetched languages for {project_path}: {languages}")
+        project["__languages"] = languages
+        yield [project]
+
+    # Helpers: Group Processing
     async def _process_single_group(
         self,
         group: dict[str, Any],
@@ -112,3 +161,79 @@ async def _process_single_group(
                     f"Fetched {len(resource_batch)} {resource_type} for group {group_id}"
                 )
                 yield resource_batch
+
+    # Helpers: File Processing and Search
+    async def _process_file(self, file: dict[str, Any], context: str) -> dict[str, Any]:
+        """Fetch full file content and parse it."""
+        file_path = file["path"]
+        project_id = file["project_id"]
+        ref = file.get("ref", "main")
+        full_content = await self.get_file_content(project_id, file_path, ref)
+        if full_content is not None:
+            try:
+                file["content"] = await anyio.to_thread.run_sync(
+                    parse_file_content, full_content, file_path, context
+                )
+            except Exception as e:
+                logger.error(f"Failed to parse {file_path} in {context}: {str(e)}")
+                file["parsed_data"] = None
+        else:
+            file["content"] = None
+        return file
+
+    async def _process_batch(
+        self, batch: list[dict[str, Any]], context: str
+    ) -> AsyncIterator[dict[str, Any]]:
+        PARSEABLE_EXTENSIONS = (".json", ".yaml", ".yml")
+        tasks = [
+            (
+                self._process_file(file, context)
+                if file.get("path", "").endswith(PARSEABLE_EXTENSIONS)
+                else asyncio.create_task(asyncio.sleep(0, result=file))
+            )
+            for file in batch
+        ]
+        for completed in asyncio.as_completed(tasks):
+            yield await completed
+
+    async def _search_in_repository(
+        self,
+        repo: str,
+        scope: str,
+        query: str,
+    ) -> AsyncIterator[list[dict[str, Any]]]:
+        """Search files in a repository using a scope and query."""
+        logger.debug(
+            f"Searching repo '{repo}' for query '{query}' with scope '{scope}'"
+        )
+        params = {"scope": scope, "search": query}
+        encoded_repo = quote(repo, safe="")
+        path = f"projects/{encoded_repo}/search"
+        async for batch in self.rest.get_paginated_resource(path, params=params):
+            if batch:
+                processed_batch = []
+                async for processed_file in self._process_batch(batch, repo):
+                    processed_batch.append(processed_file)
+                if processed_batch:
+                    yield processed_batch
+
+    async def _search_in_group(
+        self,
+        group: dict[str, Any],
+        scope: str,
+        query: str,
+    ) -> AsyncIterator[list[dict[str, Any]]]:
+        """Search files in a group using a scope and query."""
+        group_context = group.get("name", str(group["id"]))
+        group_id = group["id"]
+        params = {"scope": scope, "search": query}
+        logger.debug(f"Searching group '{group_context}' for query '{query}'")
+        async for batch in self.rest.get_paginated_group_resource(
+            group_id, "search", params
+        ):
+            if batch:
+                processed_batch = []
+                async for processed_file in self._process_batch(batch, group_context):
+                    processed_batch.append(processed_file)
+                if processed_batch:
+                    yield processed_batch

integrations/gitlab-v2/clients/rest_client.py

@@ -1,3 +1,4 @@
+import base64
 from typing import Any, AsyncIterator, Optional
 from urllib.parse import quote
 
@@ -8,7 +9,7 @@
 
 class RestClient(HTTPBaseClient):
     DEFAULT_PAGE_SIZE = 100
-    VALID_GROUP_RESOURCES = ["issues", "merge_requests", "labels"]
+    VALID_GROUP_RESOURCES = ["issues", "merge_requests", "labels", "search"]
 
     async def get_paginated_resource(
         self, resource_type: str, params: Optional[dict[str, Any]] = None
@@ -49,6 +50,20 @@ async def get_project_languages(
         path = f"projects/{encoded_project_path}/languages"
         return await self.send_api_request("GET", path, params=params or {})
 
+    async def get_file_content(
+        self, project_id: str, file_path: str, ref: str = "main"
+    ) -> Optional[str]:
+        encoded_project_id = quote(str(project_id), safe="")
+        encoded_file_path = quote(file_path, safe="")
+        path = f"projects/{encoded_project_id}/repository/files/{encoded_file_path}"
+        params = {"ref": ref}
+
+        response = await self.send_api_request("GET", path, params=params)
+        if not response:
+            return None
+
+        return base64.b64decode(response["content"]).decode("utf-8")
+
     async def _make_paginated_request(
         self,
         path: str,
@@ -57,23 +72,16 @@ async def _make_paginated_request(
     ) -> AsyncIterator[list[dict[str, Any]]]:
         page = 1
         params_dict: dict[str, Any] = params or {}
-
         while True:
             request_params = {**params_dict, "per_page": page_size, "page": page}
             logger.debug(f"Fetching page {page} from {path}")
-
             response = await self.send_api_request("GET", path, params=request_params)
-
             # HTTP API returns a list directly, or empty dict for 404
             batch: list[dict[str, Any]] = response if isinstance(response, list) else []
-
             if not batch:
                 break
-
             yield batch
-
             if len(batch) < page_size:
                 logger.debug(f"Last page reached for {path}, no more data.")
                 break
-
             page += 1

integrations/gitlab-v2/clients/utils.py

@@ -0,0 +1,47 @@
+import json
+from typing import Any, Union
+
+import yaml
+from braceexpand import braceexpand
+from loguru import logger
+
+
+def strip_recursive_prefix(pattern: str) -> str:
+    return pattern[3:] if pattern.startswith("**/") else pattern
+
+
+def convert_glob_to_gitlab_patterns(pattern: Union[str, list[str]]) -> list[str]:
+    """Converts glob patterns into GitLab-compatible patterns."""
+    if isinstance(pattern, list):
+        expanded_patterns: list[str] = []
+        for glob_pattern in pattern:
+            stripped_pattern = strip_recursive_prefix(glob_pattern)
+            expanded_patterns.extend(braceexpand(stripped_pattern))
+        return expanded_patterns
+
+    # Handle single pattern
+    stripped_pattern = strip_recursive_prefix(pattern)
+    return list(braceexpand(stripped_pattern))
+
+
+def parse_file_content(
+    content: str, file_path: str = "unknown", context: str = "unknown"
+) -> Union[str, dict[str, Any], list[Any]]:
+    """Parse file content as JSON or YAML, falling back to raw string if parsing fails."""
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError:
+        try:
+            logger.debug(f"Trying to parse file {file_path} in {context} as YAML")
+            documents = list(yaml.load_all(content, Loader=yaml.SafeLoader))
+            if not documents:
+                logger.debug(
+                    f"Failed to parse {file_path} as YAML, returning raw content"
+                )
+                return content
+            return documents if len(documents) > 1 else documents[0]
+        except yaml.YAMLError:
+            logger.debug(
+                f"Failed to parse {file_path} as JSON/YAML, returning raw content"
+            )
+            return content