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
- 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: