Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:TobikoData Sqlmesh CICD Deploy Production

From Leeroopedia


Knowledge Sources
Domains Data_Engineering, CICD
Last Updated 2026-02-07 00:00 GMT

Overview

Concrete implementation for enforcing approval gates and executing production deployments with automated validation and cleanup provided by SQLMesh.

Description

The deploy_production command, check_required_approvers command, and deploy_to_prod controller method implement the full production deployment workflow with approval enforcement. The implementation checks GitHub PR reviews against configured required approvers, validates merge readiness through GitHub's merge state status, verifies branch protection requirements are satisfied, applies the production plan when all conditions are met, optionally merges the PR using the configured merge method, invalidates the PR environment for cleanup, and updates GitHub Check Runs and PR comments throughout the process. This provides governed, automated production deployments with human oversight.

Usage

Use deploy_production and check_required_approvers as GitHub Actions workflow steps after successful PR environment creation and plan generation. These commands typically run only on PRs targeting production branches (main/master) and may be triggered automatically or via PR comment commands depending on configuration.

Code Reference

Source Location

  • Repository: sqlmesh
  • File: sqlmesh/integrations/github/cicd/command.py:L218-223, L66-71 (CLI commands), sqlmesh/integrations/github/cicd/controller.py:L764-801, L385-395 (controller methods)

Signature

# Deploy Production CLI Command
@github.command()
@click.pass_context
@cli_analytics
def deploy_production(ctx: click.Context) -> None:
    """Deploys the production environment"""
    if not _deploy_production(ctx.obj["github"]):
        raise CICDBotError(
            "Failed to deploy to production. See Pull Requests Checks for more information."
        )

# Check Required Approvers CLI Command
@github.command()
@click.pass_context
@cli_analytics
def check_required_approvers(ctx: click.Context) -> None:
    """Checks if a required approver has provided approval on the PR."""
    if not _check_required_approvers(ctx.obj["github"]):
        raise CICDBotError(
            "Required approver has not approved the PR. See Pull Requests Checks for more information."
        )

# Deploy to Prod Controller Method
def deploy_to_prod(self) -> None:
    """
    Attempts to deploy a plan to prod. If the plan is not up-to-date or has gaps then it will raise.
    """
    # If the PR is already merged then we will not deploy to prod if this event was triggered prior to the merge.
    if self._pull_request.merged and not self._event.is_pull_request_closed:
        raise CICDBotError(
            "PR is already merged and this event was triggered prior to the merge."
        )
    merge_status = self._get_merge_state_status()
    if self.bot_config.check_if_blocked_on_deploy_to_prod and merge_status.is_blocked:
        raise CICDBotError(
            "Branch protection or ruleset requirement is likely not satisfied, e.g. missing CODEOWNERS approval. "
            "Please check PR and resolve any issues. To disable this check, set `check_if_blocked_on_deploy_to_prod` to false in the bot configuration."
        )
    if merge_status.is_dirty:
        raise CICDBotError(
            "Merge commit cannot be cleanly created. Likely from a merge conflict. "
            "Please check PR and resolve any issues."
        )
    plan_summary = f"""<details>
  <summary>:ship: Prod Plan Being Applied</summary>

{self.get_plan_summary(self.prod_plan)}
</details>

"""
    if self.forward_only_plan:
        plan_summary = (
            f"{self.get_forward_only_plan_post_deployment_tip(self.prod_plan)}\n{plan_summary}"
        )

    self.update_sqlmesh_comment_info(
        value=plan_summary,
        dedup_regex=None,
    )
    self._context.apply(self.prod_plan)

# Has Required Approval Property
@property
def has_required_approval(self) -> bool:
    """
    Check if the PR has a required approver.
    """
    if not self._required_approvers or self._required_approvers_with_approval:
        logger.debug("Has required Approval")
        return True
    logger.debug("Does not have required approval")
    return False

Import

from sqlmesh.integrations.github.cicd.command import deploy_production, check_required_approvers
from sqlmesh.integrations.github.cicd.controller import GithubController

I/O Contract

Inputs

Name Type Required Description
ctx click.Context Yes Click context containing initialized GithubController in ctx.obj["github"]

Outputs

Name Type Description
None None Commands execute deployment, update GitHub Check Runs, optionally merge PR, and invalidate PR environment; raise CICDBotError if deployment fails

Usage Examples

Basic Usage

# In GitHub Actions workflow with automatic deployment:
# - name: Check Required Approvers
#   if: github.base_ref == 'main'
#   env:
#     GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
#   run: sqlmesh_cicd bot github check-required-approvers
#
# - name: Deploy to Production
#   if: github.base_ref == 'main'
#   env:
#     GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
#   run: sqlmesh_cicd bot github deploy-production

# In GitHub Actions workflow with command-based deployment:
# - name: Deploy to Production
#   if: github.event_name == 'issue_comment' && contains(github.event.comment.body, '/deploy')
#   env:
#     GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
#   run: sqlmesh_cicd bot github deploy-production

# Programmatic usage:
from sqlmesh.integrations.github.cicd.controller import (
    GithubController,
    GithubCheckStatus,
    GithubCheckConclusion
)

controller = GithubController(
    paths=["path/to/project"],
    token="github_token"
)

# Check approvals first
controller.update_required_approval_check(status=GithubCheckStatus.IN_PROGRESS)
if controller.has_required_approval:
    controller.update_required_approval_check(
        status=GithubCheckStatus.COMPLETED,
        conclusion=GithubCheckConclusion.SUCCESS
    )
else:
    controller.update_required_approval_check(
        status=GithubCheckStatus.COMPLETED,
        conclusion=GithubCheckConclusion.FAILURE
    )
    raise Exception("Required approver has not approved")

# Deploy to production
controller.update_prod_environment_check(status=GithubCheckStatus.IN_PROGRESS)
try:
    controller.deploy_to_prod()
    controller.update_prod_environment_check(
        status=GithubCheckStatus.COMPLETED,
        conclusion=GithubCheckConclusion.SUCCESS
    )

    # Merge PR if configured
    controller.try_merge_pr()

    # Clean up PR environment
    controller.try_invalidate_pr_environment()

except Exception as e:
    controller.update_prod_environment_check(
        status=GithubCheckStatus.COMPLETED,
        conclusion=GithubCheckConclusion.FAILURE
    )
    raise

# Example PR comment after deployment:
# 🤖 **SQLMesh Bot Info** 🤖
# - 👀 To **review** this PR's changes, use virtual data environment:
#   - `my_repo_123`
# <details>
#   <summary>🚢 Prod Plan Being Applied</summary>
#
# Summary of differences against `prod`:
# └── Modified Models:
#     └── Directly Modified:
#         └── db.orders (Breaking) [full refresh]
#
# Successfully applied production plan!
# </details>

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment