Skip to content

taglyatelle_api.py

Expose taglyatelle as an API.

ping()

Ping the taglyatelle API.

Source code in taglyatelle/exposition/taglyatelle_api.py 
25
26
27
28
@app.get("/taglyatelle/ping")
def ping() -> str:
    """Ping the taglyatelle API."""
    return "pong"

receipt_payload(request, webhook_event=Depends(WebhookEventModel.from_header)) async

Receipt webhook events from git providers and trigger some actions.

Parameters

request The incoming request object containing the webhook payload

webhook_event The webhook event model containing the event header

Source code in taglyatelle/exposition/taglyatelle_api.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@app.post("/taglyatelle/webhooks")
async def receipt_payload(
    request: Request,
    webhook_event: WebhookEventModel = Depends(WebhookEventModel.from_header),
):
    """
    Receipt webhook events from git providers and trigger some actions.

    Parameters
    ----------
    request
        The incoming request object containing the webhook payload

    webhook_event
        The webhook event model containing the event header
    """
    payload_body = await request.body()
    payload = json.loads(payload_body)

    git_request = GitProviderRequest.from_payload(
        payload=payload,
        provider=webhook_event.provider,
        event_type=webhook_event.event_type,
    )

    if git_request.installation_id:
        os.environ["INSTALLATION_ID"] = str(git_request.installation_id)

    provider = GitProvider(
        git_provider=git_request.provider,
        owner=git_request.repository_owner,
        repo=git_request.repository_name,
    )

    provider.set_llm_strategy(provider=str(os.getenv("LLM_PROVIDER")), model=str(os.getenv("LLM_MODEL")))

    # Call slash commands
    if git_request.event_type == "issue_comment" and git_request.action == "created":
        if git_request.comment_body and git_request.comment_body.strip().startswith("/"):
            comment_body = git_request.comment_body.strip()
            command = comment_body.split()[0][1:].lower()
            slash_command = SlashCommand(command, provider, git_request.raw_payload)
            slash_command.execute()

    # Synchronize changelog and create releases
    if git_request.event_type in ["pull_request", "merge_request"]:
        if git_request.action in ["opened", "synchronize", "reopened"]:
            if git_request.number:
                synchronize_changelog(provider=provider, pr_number=git_request.number)

        elif git_request.is_merged and git_request.base_ref == git_request.default_branch:
            if git_request.number:
                publish_release(provider=provider, pr_number=git_request.number)