Implementation:TobikoData Sqlmesh CICD Deploy Production
| 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>