tessl/pypi-python-gitlab
The python wrapper for the GitLab REST and GraphQL APIs.
—
Pending
graphql.mddocs/
GraphQL Clients
Advanced GraphQL query execution with both synchronous and asynchronous clients. Provides flexible querying capabilities and efficient data fetching for complex GitLab API interactions using GitLab's GraphQL API.
Capabilities
Synchronous GraphQL Client
class GraphQL:
"""
Synchronous GraphQL client for GitLab API queries.
Provides GraphQL query execution with retry logic, error handling,
and response processing for GitLab's GraphQL API endpoint.
"""
def __init__(
self,
url: str,
auth: tuple[str, str] | None = None,
timeout: float | None = None,
retry_transient_errors: bool = False,
**kwargs
) -> None:
"""
Initialize GraphQL client.
Parameters:
- url: GitLab GraphQL endpoint URL
- auth: Authentication tuple (token_type, token)
- timeout: Request timeout in seconds
- retry_transient_errors: Retry on transient errors
"""
def execute(
self,
query: str,
variables: dict | None = None,
**kwargs
) -> dict:
"""
Execute GraphQL query.
Parameters:
- query: GraphQL query string
- variables: Query variables dictionary
Returns:
Dictionary with query results
Raises:
GitlabGraphQLError: If query execution fails
Example:
```python
query = '''
query getProject($path: ID!) {
project(fullPath: $path) {
id
name
description
webUrl
repository {
tree {
lastCommit {
sha
message
authoredDate
}
}
}
issues(state: OPENED) {
count
edges {
node {
id
title
state
createdAt
}
}
}
}
}
'''
variables = {"path": "group/project"}
result = client.execute(query, variables)
```
"""Asynchronous GraphQL Client
class AsyncGraphQL:
"""
Asynchronous GraphQL client for GitLab API queries.
Provides async GraphQL query execution with the same capabilities
as the synchronous client but with non-blocking I/O operations.
"""
def __init__(
self,
url: str,
auth: tuple[str, str] | None = None,
timeout: float | None = None,
retry_transient_errors: bool = False,
**kwargs
) -> None:
"""
Initialize async GraphQL client.
Parameters:
- url: GitLab GraphQL endpoint URL
- auth: Authentication tuple (token_type, token)
- timeout: Request timeout in seconds
- retry_transient_errors: Retry on transient errors
"""
async def execute(
self,
query: str,
variables: dict | None = None,
**kwargs
) -> dict:
"""
Execute GraphQL query asynchronously.
Parameters:
- query: GraphQL query string
- variables: Query variables dictionary
Returns:
Dictionary with query results
Raises:
GitlabGraphQLError: If query execution fails
Example:
```python
import asyncio
async def main():
query = '''
query getUserProjects($userId: UserID!) {
user(id: $userId) {
id
name
username
projectMemberships {
edges {
node {
project {
id
name
visibility
lastActivityAt
}
accessLevel {
integerValue
}
}
}
}
}
}
'''
variables = {"userId": "gid://gitlab/User/123"}
result = await client.execute(query, variables)
asyncio.run(main())
```
"""Client Creation and Configuration
# Create GraphQL clients from Gitlab instance
@property
def graphql(self) -> GraphQL:
"""
Get synchronous GraphQL client configured for this GitLab instance.
Returns:
Configured GraphQL client
Example:
```python
gl = gitlab.Gitlab("https://gitlab.example.com", private_token="token")
gql = gl.graphql
query = "{ currentUser { id name } }"
result = gql.execute(query)
```
"""
@property
def async_graphql(self) -> AsyncGraphQL:
"""
Get asynchronous GraphQL client configured for this GitLab instance.
Returns:
Configured AsyncGraphQL client
Example:
```python
gl = gitlab.Gitlab("https://gitlab.example.com", private_token="token")
async_gql = gl.async_graphql
async def get_user():
query = "{ currentUser { id name } }"
result = await async_gql.execute(query)
return result
```
"""GraphQL Schema and Common Queries
# Common GraphQL query patterns for GitLab
# Project Information Query
PROJECT_QUERY = """
query getProject($fullPath: ID!) {
project(fullPath: $fullPath) {
id
name
path
fullPath
description
visibility
webUrl
sshUrlToRepo
httpUrlToRepo
defaultBranch
repository {
exists
empty
rootRef
}
statistics {
commitCount
storageSize
repositorySize
wikiSize
lfsObjectsSize
jobArtifactsSize
packagesSize
snippetsSize
}
lastActivityAt
createdAt
updatedAt
}
}
"""
# Issues Query
ISSUES_QUERY = """
query getProjectIssues($fullPath: ID!, $state: IssuableState, $first: Int) {
project(fullPath: $fullPath) {
issues(state: $state, first: $first) {
count
pageInfo {
hasNextPage
hasPreviousPage
startCursor
endCursor
}
edges {
node {
id
iid
title
description
state
createdAt
updatedAt
closedAt
author {
id
name
username
avatarUrl
}
assignees {
edges {
node {
id
name
username
}
}
}
labels {
edges {
node {
id
title
color
description
}
}
}
milestone {
id
title
state
dueDate
}
webUrl
upvotes
downvotes
userNotesCount
confidential
discussionLocked
dueDate
timeEstimate
totalTimeSpent
humanTimeEstimate
humanTotalTimeSpent
}
}
}
}
}
"""
# Merge Requests Query
MERGE_REQUESTS_QUERY = """
query getProjectMergeRequests($fullPath: ID!, $state: MergeRequestState, $first: Int) {
project(fullPath: $fullPath) {
mergeRequests(state: $state, first: $first) {
count
pageInfo {
hasNextPage
hasPreviousPage
startCursor
endCursor
}
edges {
node {
id
iid
title
description
state
createdAt
updatedAt
mergedAt
closedAt
sourceBranch
targetBranch
author {
id
name
username
avatarUrl
}
assignees {
edges {
node {
id
name
username
}
}
}
reviewers {
edges {
node {
id
name
username
}
}
}
labels {
edges {
node {
id
title
color
}
}
}
milestone {
id
title
state
}
webUrl
upvotes
downvotes
userNotesCount
shouldRemoveSourceBranch
forceRemoveSourceBranch
squash
mergeable
mergeStatus
draft
workInProgress
conflicts
divergedCommitsCount
rebaseInProgress
defaultMergeCommitMessage
mergeCommitSha
squashCommitSha
diffHeadSha
diffBaseSha
reference
taskCompletionStatus {
count
completedCount
}
}
}
}
}
}
"""
# Users Query
USERS_QUERY = """
query getUsers($search: String, $first: Int) {
users(search: $search, first: $first) {
count
pageInfo {
hasNextPage
hasPreviousPage
startCursor
endCursor
}
edges {
node {
id
name
username
email
avatarUrl
webUrl
bot
state
createdAt
lastActivityOn
location
publicEmail
skype
linkedin
twitter
websiteUrl
organization
jobTitle
bio
workInformation
localTime
pronouns
bot
followers {
count
}
following {
count
}
groupMemberships {
count
}
projectMemberships {
count
}
starredProjects {
count
}
status {
availability
emoji
message
}
}
}
}
}
"""
# Groups Query
GROUPS_QUERY = """
query getGroups($search: String, $first: Int) {
groups(search: $search, first: $first) {
count
pageInfo {
hasNextPage
hasPreviousPage
startCursor
endCursor
}
edges {
node {
id
name
path
fullName
fullPath
description
visibility
webUrl
avatarUrl
parent {
id
name
fullPath
}
projects {
count
}
descendantGroups {
count
}
createdAt
requestAccessEnabled
requireTwoFactorAuthentication
twoFactorGracePeriod
autoDevopsEnabled
emailsDisabled
mentionsDisabled
lfsEnabled
shareWithGroupLock
projectCreationLevel
subgroupCreationLevel
defaultBranchProtection
repositoryStorageLimit
}
}
}
}
"""Usage Examples
import gitlab
import asyncio
# Initialize GitLab client
gl = gitlab.Gitlab("https://gitlab.example.com", private_token="your-token")
# Synchronous GraphQL usage
def sync_graphql_example():
gql = gl.graphql
# Simple current user query
user_query = """
{
currentUser {
id
name
username
email
avatarUrl
}
}
"""
result = gql.execute(user_query)
user = result['data']['currentUser']
print(f"Current user: {user['name']} ({user['username']})")
# Project query with variables
project_query = """
query getProject($path: ID!) {
project(fullPath: $path) {
id
name
description
visibility
webUrl
issues(state: OPENED, first: 5) {
count
edges {
node {
id
title
author {
name
}
createdAt
}
}
}
mergeRequests(state: OPENED, first: 5) {
count
edges {
node {
id
title
author {
name
}
sourceBranch
targetBranch
}
}
}
}
}
"""
variables = {"path": "group/project-name"}
result = gql.execute(project_query, variables)
project = result['data']['project']
print(f"Project: {project['name']}")
print(f"Open issues: {project['issues']['count']}")
print(f"Open MRs: {project['mergeRequests']['count']}")
# Asynchronous GraphQL usage
async def async_graphql_example():
async_gql = gl.async_graphql
# Batch multiple queries
queries = [
{
"query": """
query getProject($path: ID!) {
project(fullPath: $path) {
id
name
statistics {
commitCount
storageSize
}
}
}
""",
"variables": {"path": "group/project1"}
},
{
"query": """
query getProject($path: ID!) {
project(fullPath: $path) {
id
name
statistics {
commitCount
storageSize
}
}
}
""",
"variables": {"path": "group/project2"}
}
]
# Execute queries concurrently
tasks = [
async_gql.execute(q["query"], q["variables"])
for q in queries
]
results = await asyncio.gather(*tasks)
for result in results:
project = result['data']['project']
stats = project['statistics']
print(f"{project['name']}: {stats['commitCount']} commits, {stats['storageSize']} bytes")
# Complex pagination example
def paginated_graphql_query():
gql = gl.graphql
query = """
query getProjectIssues($path: ID!, $after: String) {
project(fullPath: $path) {
issues(first: 50, after: $after) {
pageInfo {
hasNextPage
endCursor
}
edges {
node {
id
title
state
createdAt
author {
name
}
}
}
}
}
}
"""
all_issues = []
variables = {"path": "group/project"}
has_next_page = True
while has_next_page:
result = gql.execute(query, variables)
issues_data = result['data']['project']['issues']
# Collect issues
issues = [edge['node'] for edge in issues_data['edges']]
all_issues.extend(issues)
# Check for next page
page_info = issues_data['pageInfo']
has_next_page = page_info['hasNextPage']
if has_next_page:
variables['after'] = page_info['endCursor']
print(f"Retrieved {len(all_issues)} issues total")
return all_issues
# Mutation example (create issue)
def create_issue_mutation():
gql = gl.graphql
mutation = """
mutation createIssue($input: CreateIssueInput!) {
createIssue(input: $input) {
issue {
id
iid
title
description
state
webUrl
}
errors
}
}
"""
variables = {
"input": {
"projectPath": "group/project",
"title": "New issue via GraphQL",
"description": "This issue was created using GraphQL mutation",
"confidential": False
}
}
result = gql.execute(mutation, variables)
if result['data']['createIssue']['errors']:
print(f"Errors: {result['data']['createIssue']['errors']}")
else:
issue = result['data']['createIssue']['issue']
print(f"Created issue #{issue['iid']}: {issue['title']}")
print(f"URL: {issue['webUrl']}")
# Run examples
sync_graphql_example()
asyncio.run(async_graphql_example())
paginated_issues = paginated_graphql_query()
create_issue_mutation()Error Handling
class GitlabGraphQLError(GitlabError):
"""Raised when GraphQL query execution fails."""
passExample error handling:
try:
result = gql.execute(query, variables)
# Check for GraphQL errors in response
if 'errors' in result:
for error in result['errors']:
print(f"GraphQL Error: {error['message']}")
if 'locations' in error:
print(f"Location: {error['locations']}")
else:
# Process successful result
data = result['data']
except gitlab.GitlabGraphQLError as e:
print(f"GraphQL execution failed: {e}")
except gitlab.GitlabAuthenticationError:
print("Authentication failed - check your token")
except gitlab.GitlabConnectionError:
print("Connection failed - check your URL")GraphQL vs REST API
GraphQL provides several advantages over the REST API:
- Single Request: Fetch related data in one query instead of multiple REST calls
- Flexible Data Selection: Request only the fields you need
- Strong Type System: Built-in validation and introspection
- Real-time Subscriptions: Subscribe to data changes (GitLab Premium feature)
- Efficient Pagination: Cursor-based pagination with connection patterns
- Reduced Over-fetching: Get exactly the data you need
Use GraphQL when:
- You need data from multiple related resources
- You want to minimize network requests
- You need complex filtering or nested data
- You're building data-heavy applications
- You want strongly-typed API interactions
Use REST API when:
- You need CRUD operations on single resources
- You're performing administrative actions
- You need features not yet available in GraphQL
- You're working with legacy systems expecting REST endpoints
Install with Tessl CLI
npx tessl i tessl/pypi-python-gitlab