gidgetlab.aiohttp — aiohttp support

class gidgetlab.aiohttp.GitLabAPI(session, *args, **kwargs)

An implementation of gidgetlab.abc.GitLabAPI using aiohttp.

Typical usage will be:

import aiohttp
import gidgetlab.aiohttp


async with aiohttp.ClientSession() as session:
    gl = gidgetlab.aiohttp.GitLabAPI(session, requester,
                                     access_token=access_token)
    # Make your requests, e.g. ...
    data = await gl.getitem("/templates/licenses/MIT")

This module even provides a GitLabBot class to create simple aiohttp web server that handles GitLab webhooks.

class gidgetlab.aiohttp.GitLabBot(requester, *, secret=None, access_token=None, cache=None, ssl_context=None, wait_consistency=True, **kwargs)

A GitLabBot is an aiohttp web server that handles GitLab webhooks requests

If not given in arguments the webhook secret and access token are retrieved from the environment variables GL_SECRET and GL_ACCESS_TOKEN.

If not given, the cache is set to cachetools.LRUCache(maxsize=500).

If provided, the ssl.SSLContext from ssl_context will be passed to the underlying aiohttp.ClientSession.

By default, the bot waits 1 second before to process the webhook to give GitLab some time to reach internal consistency. Depending on your use case (if webhooks are used to interact with an external service for example), you might want to disable this 1 second pause by setting wait_consistency to False.

The extra kwargs are passed to the GitLabAPI instance and can be used to set a specific url and api_version.

Typical usage is:

from gidgetlab.aiohttp import GitLabBot

bot = GitLabBot("username")

# Register a callack for a webhook event
@bot.router.register("Issue Hook", action="open")
async def issue_opened_event(event, gl, *args, **kwargs):
    url = f"/projects/{event.project_id}/issues/{event.object_attributes['iid']}/notes"
    message = f"Thanks for the report @{event.data['user']['username']}! I will look into it ASAP! (I'm a bot)."
    await gl.post(url, data={"body": message})

bot.run()

Several routers can also be registered afterwards:

import gidgetlab.routing
from gidgetlab.aiohttp import GitLabBot

first_router = gidgetlab.routing.Router()
second_router = gidgetlab.routing.Router()

@first_router.register("Issue Hook", action="open")
async def issue_opened_event(event, gl, *args, **kwargs):
    ...

@second_router.register("Push Hook")
async def push_event(event, gl, *args, **kwargs):
    ...

bot = GitLabBot("username")
bot.register_routers(first_router, second_router)
bot.run()
property router: Router

The bot gidgetlab.routing.Router instance that routes webhooks events callbacks

register_routers(*routers)

Instantiate the bot router from the given routers

Return type:

None

async health_handler(request)

Handler to check the health of the bot

Return ‘Bot OK’

Return type:

Response

async webhook_handler(request)

Handler that processes GitLab webhook requests

Return type:

Response

run(**kwargs)

Run the bot web server

All keyword arguments are passed to the aiohttp.web.run_app() function.

Return type:

None