mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-16 23:41:43 +00:00
215bf03373
Narrows reportAny / reportExplicitAny hot spots in provider transformations, proxy endpoints, integrations and secret managers by introducing TypedDicts, Protocols and object-typed boundaries instead of Any, then ratchets the budget ceilings down to match. reportAny 14765 -> 14076, reportExplicitAny 4493 -> 4128, ANN401 387 -> 307
359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""
|
|
GitLab API client for fetching files from GitLab repositories.
|
|
Now supports selecting a tag via `config["tag"]`; falls back to branch ("main").
|
|
"""
|
|
|
|
import base64
|
|
from collections.abc import Mapping, Sequence
|
|
from typing import Any, Final, Protocol, TypedDict
|
|
from urllib.parse import quote
|
|
|
|
from typing_extensions import ReadOnly
|
|
|
|
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
|
|
|
|
|
class GitLabFilePayload(TypedDict, total=False):
|
|
"""A repository-files API entry."""
|
|
|
|
content: ReadOnly[str]
|
|
encoding: ReadOnly[str]
|
|
|
|
|
|
class GitLabTreeEntry(TypedDict, total=False):
|
|
"""A repository-tree API entry."""
|
|
|
|
path: ReadOnly[str]
|
|
type: ReadOnly[str]
|
|
|
|
|
|
class GitLabBranch(TypedDict, total=False):
|
|
"""A repository-branches API entry."""
|
|
|
|
name: ReadOnly[str]
|
|
type: ReadOnly[str]
|
|
|
|
|
|
class GitLabFileMetadata(TypedDict):
|
|
"""The response headers a raw file request exposes as metadata."""
|
|
|
|
content_type: ReadOnly[str | None]
|
|
content_length: ReadOnly[str | None]
|
|
last_modified: ReadOnly[str | None]
|
|
|
|
|
|
class _FileJsonResponse(Protocol):
|
|
def json(self) -> GitLabFilePayload: ...
|
|
|
|
|
|
class _TreeJsonResponse(Protocol):
|
|
def json(self) -> Sequence[GitLabTreeEntry] | None: ...
|
|
|
|
|
|
class _ProjectJsonResponse(Protocol):
|
|
def json(self) -> Mapping[str, object]: ...
|
|
|
|
|
|
class _BranchesJsonResponse(Protocol):
|
|
def json(self) -> Sequence[GitLabBranch] | None: ...
|
|
|
|
|
|
def _file_payload(resp: _FileJsonResponse) -> GitLabFilePayload:
|
|
"""The JSON body of a repository-files response."""
|
|
return resp.json()
|
|
|
|
|
|
def _tree_entries(resp: _TreeJsonResponse) -> Sequence[GitLabTreeEntry]:
|
|
"""The entries of a repository-tree response."""
|
|
return resp.json() or []
|
|
|
|
|
|
def _project_info(resp: _ProjectJsonResponse) -> Mapping[str, object]:
|
|
"""The JSON body of a project response."""
|
|
return resp.json()
|
|
|
|
|
|
def _branch_entries(resp: _BranchesJsonResponse) -> Sequence[GitLabBranch] | None:
|
|
"""The JSON body of a repository-branches response."""
|
|
return resp.json()
|
|
|
|
|
|
class GitLabClient:
|
|
"""
|
|
Client for interacting with the GitLab API to fetch files.
|
|
|
|
Supports:
|
|
- Authentication with personal/access tokens or OAuth bearer tokens
|
|
- Fetching file contents from repositories (raw endpoint with JSON fallback)
|
|
- Namespace/project path or numeric project ID addressing
|
|
- Ref selection via tag (preferred) or branch (default "main")
|
|
- Directory listing via the repository tree API
|
|
"""
|
|
|
|
def __init__(self, config: dict[str, Any]):
|
|
"""
|
|
Initialize the GitLab client.
|
|
|
|
Args:
|
|
config: Dictionary containing:
|
|
- project: Project path ("group/subgroup/repo") or numeric project ID (str|int) [required]
|
|
- access_token: GitLab personal/access token or OAuth token [required] (str)
|
|
- auth_method: 'token' (default; sends Private-Token) or 'oauth' (Authorization: Bearer)
|
|
- tag: Tag name to fetch from (takes precedence over branch if provided)
|
|
- branch: Branch to fetch from (default: "main")
|
|
- base_url: Base GitLab API URL (default: "https://gitlab.com/api/v4")
|
|
"""
|
|
project: Final = config.get("project")
|
|
access_token: Final = config.get("access_token")
|
|
if project is None or access_token is None:
|
|
raise ValueError("project and access_token are required")
|
|
|
|
self.project: str | int = project
|
|
self.access_token: str = str(access_token)
|
|
self.auth_method: str = config.get("auth_method", "token") # 'token' or 'oauth'
|
|
self.branch = config.get("branch", None)
|
|
if not self.branch:
|
|
self.branch = "main"
|
|
self.tag = config.get("tag")
|
|
self.base_url: str = config.get("base_url", "https://gitlab.com/api/v4")
|
|
|
|
if not all([self.project, self.access_token]):
|
|
raise ValueError("project and access_token are required")
|
|
|
|
# Effective ref: prefer tag if provided, else branch ("main")
|
|
self.ref = str(self.tag or self.branch)
|
|
|
|
# Build headers
|
|
self.headers = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
}
|
|
if self.auth_method == "oauth":
|
|
self.headers["Authorization"] = f"Bearer {self.access_token}"
|
|
else:
|
|
# Default GitLab token header
|
|
self.headers["Private-Token"] = self.access_token
|
|
|
|
# Project identifier must be URL-encoded (slashes become %2F)
|
|
self._project_enc = quote(str(self.project), safe="")
|
|
|
|
# HTTP handler
|
|
self.http_handler = HTTPHandler()
|
|
|
|
# ------------------------
|
|
# Core helpers
|
|
# ------------------------
|
|
|
|
def _file_raw_url(self, file_path: str, *, ref: str | None = None) -> str:
|
|
file_enc: Final = quote(file_path, safe="")
|
|
ref_q: Final = quote(ref or self.ref, safe="")
|
|
return f"{self.base_url}/projects/{self._project_enc}/repository/files/{file_enc}/raw?ref={ref_q}"
|
|
|
|
def _file_json_url(self, file_path: str, *, ref: str | None = None) -> str:
|
|
file_enc: Final = quote(file_path, safe="")
|
|
ref_q: Final = quote(ref or self.ref, safe="")
|
|
return f"{self.base_url}/projects/{self._project_enc}/repository/files/{file_enc}?ref={ref_q}"
|
|
|
|
def _tree_url(
|
|
self,
|
|
directory_path: str = "",
|
|
recursive: bool = False,
|
|
*,
|
|
ref: str | None = None,
|
|
) -> str:
|
|
path_q: Final = f"&path={quote(directory_path, safe='')}" if directory_path else ""
|
|
rec_q: Final = "&recursive=true" if recursive else ""
|
|
ref_q: Final = quote(ref or self.ref, safe="")
|
|
return f"{self.base_url}/projects/{self._project_enc}/repository/tree?ref={ref_q}{path_q}{rec_q}"
|
|
|
|
# ------------------------
|
|
# Public API
|
|
# ------------------------
|
|
|
|
def set_ref(self, ref: str) -> None:
|
|
"""Override the default ref (tag/branch) for subsequent calls."""
|
|
if not ref:
|
|
raise ValueError("ref must be a non-empty string")
|
|
self.ref = ref
|
|
|
|
def get_file_content(self, file_path: str, *, ref: str | None = None) -> str | None:
|
|
"""
|
|
Fetch the content of a file from the GitLab repository at the given ref
|
|
(tag, branch, or commit SHA). If `ref` is None, uses self.ref.
|
|
|
|
Strategy:
|
|
1) Try the RAW endpoint (returns bytes of the file)
|
|
2) Fallback to the JSON endpoint (returns base64-encoded content)
|
|
|
|
Returns:
|
|
File content as UTF-8 string, or None if file not found.
|
|
"""
|
|
raw_url: Final = self._file_raw_url(file_path, ref=ref)
|
|
|
|
try:
|
|
resp: Final = self.http_handler.get(raw_url, headers=self.headers)
|
|
if resp.status_code == 404:
|
|
# Fallback to JSON endpoint
|
|
return self._get_file_content_via_json(file_path, ref=ref)
|
|
resp.raise_for_status()
|
|
|
|
ctype: Final = (resp.headers.get("content-type") or "").lower()
|
|
if ctype.startswith("text/") or "charset=" in ctype or ctype.startswith("application/json"):
|
|
return resp.text
|
|
try:
|
|
return resp.content.decode("utf-8")
|
|
except Exception:
|
|
return resp.content.decode("utf-8", errors="replace")
|
|
|
|
except Exception as e:
|
|
status: Final = getattr(getattr(e, "response", None), "status_code", None)
|
|
if status == 404:
|
|
return None
|
|
if status == 403:
|
|
raise Exception(
|
|
f"Access denied to file '{file_path}'. Check your GitLab permissions for project '{self.project}'."
|
|
)
|
|
if status == 401:
|
|
raise Exception("Authentication failed. Check your GitLab token and auth_method.")
|
|
raise Exception(f"Failed to fetch file '{file_path}': {e}")
|
|
|
|
def _get_file_content_via_json(self, file_path: str, *, ref: str | None = None) -> str | None:
|
|
"""
|
|
Fallback for get_file_content(): use the JSON file API which returns base64 content.
|
|
"""
|
|
json_url: Final = self._file_json_url(file_path, ref=ref)
|
|
try:
|
|
resp: Final = self.http_handler.get(json_url, headers=self.headers)
|
|
if resp.status_code == 404:
|
|
return None
|
|
resp.raise_for_status()
|
|
data: Final = _file_payload(resp)
|
|
content: Final = data.get("content")
|
|
encoding: Final = data.get("encoding", "")
|
|
if content and encoding == "base64":
|
|
try:
|
|
return base64.b64decode(content).decode("utf-8")
|
|
except Exception:
|
|
return base64.b64decode(content).decode("utf-8", errors="replace")
|
|
return content
|
|
except Exception as e:
|
|
status: Final = getattr(getattr(e, "response", None), "status_code", None)
|
|
if status == 404:
|
|
return None
|
|
if status == 403:
|
|
raise Exception(
|
|
f"Access denied to file '{file_path}'. Check your GitLab permissions for project '{self.project}'."
|
|
)
|
|
if status == 401:
|
|
raise Exception("Authentication failed. Check your GitLab token and auth_method.")
|
|
raise Exception(f"Failed to fetch file '{file_path}' via JSON endpoint: {e}")
|
|
|
|
def list_files(
|
|
self,
|
|
directory_path: str = "",
|
|
file_extension: str = ".prompt",
|
|
recursive: bool = False,
|
|
*,
|
|
ref: str | None = None,
|
|
) -> list[str]:
|
|
"""
|
|
List files in a directory with a specific extension using the repository tree API.
|
|
|
|
Args:
|
|
directory_path: Directory path in the repository (empty for repo root)
|
|
file_extension: File extension to filter by (default: .prompt)
|
|
recursive: If True, traverses subdirectories
|
|
ref: Optional override (tag/branch/SHA). Defaults to self.ref.
|
|
|
|
Returns:
|
|
List of file paths (relative to repo root)
|
|
"""
|
|
url: Final = self._tree_url(directory_path, recursive=recursive, ref=ref)
|
|
|
|
try:
|
|
resp: Final = self.http_handler.get(url, headers=self.headers)
|
|
if resp.status_code == 404:
|
|
return []
|
|
resp.raise_for_status()
|
|
|
|
data: Final = _tree_entries(resp)
|
|
files: Final[list[str]] = []
|
|
for item in data:
|
|
if item.get("type") == "blob":
|
|
file_path = item.get("path", "")
|
|
if not file_extension or file_path.endswith(file_extension):
|
|
files.append(file_path)
|
|
return files
|
|
|
|
except Exception as e:
|
|
status: Final = getattr(getattr(e, "response", None), "status_code", None)
|
|
if status == 404:
|
|
return []
|
|
if status == 403:
|
|
raise Exception(
|
|
f"Access denied to directory '{directory_path}'. Check your GitLab permissions for project '{self.project}'."
|
|
)
|
|
if status == 401:
|
|
raise Exception("Authentication failed. Check your GitLab token and auth_method.")
|
|
raise Exception(f"Failed to list files in '{directory_path}': {e}")
|
|
|
|
def get_repository_info(self) -> Mapping[str, object]:
|
|
"""Get information about the project/repository."""
|
|
url: Final = f"{self.base_url}/projects/{self._project_enc}"
|
|
try:
|
|
resp: Final = self.http_handler.get(url, headers=self.headers)
|
|
resp.raise_for_status()
|
|
return _project_info(resp)
|
|
except Exception as e:
|
|
raise Exception(f"Failed to get repository info: {e}")
|
|
|
|
def test_connection(self) -> bool:
|
|
"""Test the connection to the GitLab project."""
|
|
try:
|
|
self.get_repository_info()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def get_branches(self) -> list[GitLabBranch]:
|
|
"""Get list of branches in the repository."""
|
|
url: Final = f"{self.base_url}/projects/{self._project_enc}/repository/branches"
|
|
try:
|
|
resp: Final = self.http_handler.get(url, headers=self.headers)
|
|
resp.raise_for_status()
|
|
data: Final = _branch_entries(resp)
|
|
return data if isinstance(data, list) else []
|
|
except Exception as e:
|
|
raise Exception(f"Failed to get branches: {e}")
|
|
|
|
def get_file_metadata(self, file_path: str, *, ref: str | None = None) -> GitLabFileMetadata | None:
|
|
"""
|
|
Get minimal metadata about a file via RAW endpoint headers at a given ref.
|
|
|
|
Args:
|
|
file_path: Path to the file in the repository.
|
|
ref: Optional override (tag/branch/SHA). Defaults to self.ref.
|
|
"""
|
|
url: Final = self._file_raw_url(file_path, ref=ref)
|
|
try:
|
|
headers: Final = dict(self.headers)
|
|
headers["Range"] = "bytes=0-0"
|
|
resp: Final = self.http_handler.get(url, headers=headers)
|
|
if resp.status_code == 404:
|
|
return None
|
|
resp.raise_for_status()
|
|
return {
|
|
"content_type": resp.headers.get("content-type"),
|
|
"content_length": resp.headers.get("content-length"),
|
|
"last_modified": resp.headers.get("last-modified"),
|
|
}
|
|
except Exception as e:
|
|
status: Final = getattr(getattr(e, "response", None), "status_code", None)
|
|
if status == 404:
|
|
return None
|
|
raise Exception(f"Failed to get file metadata for '{file_path}': {e}")
|
|
|
|
def close(self):
|
|
"""Close the HTTP handler to free resources."""
|
|
if hasattr(self, "http_handler"):
|
|
self.http_handler.close()
|