There’s a moment every developer recognizes.

You open a pull request.
You scroll.
You sigh.

At first it’s small:

“Oh, someone forgot FORCE on a view.”

Then it grows:

“Why does this Flyway script sort before last week’s migration?”

And before you know it, you’re deep inside an APEX export wondering why a page region is querying tables directly, bypassing every architectural rule you ever agreed on.

This is the story of how we stopped reviewing those mistakes and started hunting them.

With a snake. 🐍

The snake isn’t smart. It doesn’t understand architecture diagrams, product roadmaps or how late it is on a Friday. It only understands “this line changed” and “you promised not to do that here”. That turns all those fuzzy “we all know this” rules into something the CI can actually enforce.

The boring heart of it: the rule base

Everything starts with one small contract. No frameworks. No AI magic. Just a base class that all rules inherit from.

"""
DATA STRUCTURES - PL/SQL RECORD TYPE equivalents
Immutable records for changes and violations
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path

@dataclass(frozen=True, slots=True)
class Change:
    file: Path
    line: int
    content: str

@dataclass(frozen=True, slots=True)
class Finding:
    file: Path
    line: int
    message: str
    rule: str
    severity: str = "ERROR"

"""
BASERULE - Template all rules inherit from (PL/SQL package spec equivalent)
"""
from __future__ import annotations
from pathlib import Path
from typing import List
from dataclasses import dataclass

from models import Finding

class BaseRule:
    def __init__(self) -> None:
        self._violations: List[Finding] = []

    def on_file(self, file_path: Path) -> None:
        """Called once per file - override in your rule"""
        pass

    def on_line(self, file_path: Path, line_no: int, content: str) -> None:
        """Called for every changed line - put logic here"""
        pass

    def violation(self, file_path: Path, line_no: int, message: str) -> None:
        """Record violation (no RAISE, just collect)"""
        self._violations.append(
            Finding(
                file=file_path,
                line=line_no,
                message=message,
                rule=self.__class__.__name__,
            )
        )

    def finalize(self) -> list[Finding]:
        """Return all collected violations"""
        return self._violations

Each rule does one thing: inspect input and emit structured violations. Nothing more.

Every violation is captured as plain JSON file, line, message and rule name. That structure is the contract. Once a violation is represented as data instead of a print statement, it becomes portable. It can be posted to Bitbucket, pushed to another Git host, sent to a REST endpoint, or stored for reporting.

The base rule class exists to enforce that structure. It standardizes how violations are recorded so the engine and the transport layer don’t need to know anything about the rule’s internal logic.

Rules stay small. The output stays predictable. And predictable output is what makes automation possible.

Rule #1: versioned filenames don’t get creative

Everyone knows the format.
Nobody types it right at 23:47 on a Friday.

"""
RULE #1: FLYWAY FILENAMES - V20260226__description.sql format
"""
import re

from pathlib import Path
from base import BaseRule

class FlywayFilenameRule(BaseRule):
    _FILENAME_PATTERN = re.compile(r"^V\d{12}__[a-z0-9_]+\.sql$", re.IGNORECASE)

    def on_file(self, file_path: Path) -> None:
        filename = file_path.name
        if not self._FILENAME_PATTERN.match(filename):
            self.violation(
                file_path=file_path,
                line_no=1,
                message=(
                    f"Invalid Flyway filename. "
                    f"Expected VYYYYMMDDHHMM__<description>.sql. "
                    f"Found '{filename}'."
                ),
            )

The effect was immediate and slightly embarrassing, but very satisfying…
Turns out the rule caught more mistakes than our reviews ever did.

Rule #2: views must behave themselves

The next classic: views without FORCE or views whose name doesn’t match the file.

"""
RULE #2: VIEWS NEED FORCE + filename match (.vw files only)
"""
import re

from pathlib import Path
from base import BaseRule

class ViewForceRule(BaseRule):
    _VIEW_PATTERN = re.compile(
        r"create\s+(or\s+replace\s+)?(force\s+)?view\s+(\w+)",
        re.IGNORECASE,
    )

    def on_line(self, file_path: Path, line_no: int, content: str) -> None:
        if file_path.suffix != ".vw":
            return

        match = self._VIEW_PATTERN.search(content)
        if not match:
            return

        if match.group(2) is None:
            self.violation(
                file_path=file_path,
                line_no=line_no,
                message="View must include FORCE and match the filename.",
            )
            return

        view_name = match.group(3)

        expected = file_path.stem
        if expected.startswith("R__"):
            expected = expected[3:]

        if view_name.lower() != expected.lower():
            self.violation(
                file_path=file_path,
                line_no=line_no,
                message=(
                    f"View name '{view_name}' does not match "
                    f"filename '{expected}'."
                ),
            )

This rule operates purely at line level. It scans each line for a CREATE VIEW statement using a single regular expression that captures both the optional FORCE keyword and the declared view name.

Because the rule first checks whether the current line actually contains a CREATE VIEW statement, it automatically skips file headers, comment blocks, and empty lines.

Git diff–based scanning

We teach the snake one trick: only look at the diff.
That keeps the signal high, runtime low.

The diff output is streamed line by line and parsed incrementally. No full file loading. No repository-wide traversal. Just a lightweight parser that translates unified diff output into a simple stream of {file, line, content} objects.

"""
GIT DIFF PARSER - Process only changed lines (+ prefixed)
Like parsing DBMS_OUTPUT lines but for git diff
"""
import re
import subprocess

from pathlib import Path
from typing import Iterator
from models import Change

_HUNK_PATTERN = re.compile(r"\+(\d+)")

def parse_diff(base_branch: str) -> Iterator[Change]:
    """Parse git diff, yield only changed lines as Change objects"""
    cmd = ["git", "diff", "--unified=0", f"origin/{base_branch}...HEAD"]
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)

    current_file: Path | None = None
    new_line: int | None = None

    for line in result.stdout:
        if line.startswith("+++ b/"):
            current_file = Path(line[6:].strip())
        elif line.startswith("@@"):
            match = _HUNK_PATTERN.search(line)
            if match:
                new_line = int(match.group(1))
        elif line.startswith("+") and not line.startswith("+++"):
            if current_file is None or new_line is None:
                continue
            yield Change(file=current_file, line=new_line, content=line[1:].rstrip())
            new_line += 1

Bitbucket: comments where it actually hurts

Once the engine has opinions, getting them into a pull request is just plumbing. In our case the pipe ends in Bitbucket, but the JSON we send is boring enough that GitHub or GitLab could pretend to be Bitbucket for a day.

"""
BITBUCKET COMMENTS - Inline PR comments via REST API
"""
import requests
import os

from typing import Final
from models import Finding

API: Final[str] = (
    f"https://api.bitbucket.org/2.0/repositories/"
    f"{os.environ['BITBUCKET_WORKSPACE']}/"
    f"{os.environ['BITBUCKET_REPO_SLUG']}/"
    f"pullrequests/{os.environ['BITBUCKET_PR_ID']}/comments"
)

AUTH: Final[tuple[str, str]] = (os.environ["BOT_USER"], os.environ["BOT_APP_PASSWORD"])

def post_comment(finding: Finding) -> None:
    """Post Finding as inline PR comment"""
    payload = {
        "content": {"raw": finding.message},
        "inline": {"path": str(finding.file), "to": finding.line},
    }
    response = requests.post(API, json=payload, auth=AUTH, timeout=10)
    response.raise_for_status()

If the snake bites, it bites exactly where you messed up.

Putting everything together: release the snake

Now the engine becomes almost boring.

"""
RULE ENGINE - Coordinates all rules through changes
"""
from typing import Iterable, Iterator, Sequence
from pathlib import Path
from base import BaseRule
from models import Change, Finding

class RuleEngine:
    def __init__(self, rules: Sequence[BaseRule]) -> None:
        """Main engine - loops changes through all rules"""
        self._rules = list(rules)
        self._seen_files: set[Path] = set()

    def run(self, changes: Iterable[Change]) -> None:
        """Process all changes through all rules"""
        for change in changes:
            if change.file not in self._seen_files:
                for rule in self._rules:
                    rule.on_file(change.file)
                self._seen_files.add(change.file)

            for rule in self._rules:
                rule.on_line(change.file, change.line, change.content)

    def findings(self) -> Iterator[Finding]:
        """Yield all violations from all rules"""
        for rule in self._rules:
            yield from rule.finalize()

"""
MAIN ENTRY - Bitbucket Pipelines CI script
Returns 1 on errors (fails pipeline)
"""
import os
import sys

from engine import RuleEngine
from diff_parser import parse_diff
from bitbucket import post_comment
from rules.view_force import ViewForceRule
from rules.flyway_filename import FlywayFilenameRule

def main() -> int:
    """CI entry point - fail pipeline on violations"""
    base_branch = os.environ["BITBUCKET_PR_DESTINATION_BRANCH"]

    engine = RuleEngine([
        ViewForceRule(),
        FlywayFilenameRule(),
    ])

    engine.run(parse_diff(base_branch))

    has_error = False
    for finding in engine.findings():
        post_comment(finding)
        if finding.severity == "ERROR":
            has_error = True

    return 1 if has_error else 0

if __name__ == "__main__":
    sys.exit(main())

If you want to try this yourself, don’t start with a grand vision. Pick one tiny annoyance you’re always commenting on, turn it into a rule, and let the snake do the nagging next time.

For now it lives in our pipeline, happily chewing through diffs so reviewers can focus on the parts that actually require a brain.

Looking ahead: APEXLang changes the game

Once Oracle finally releases APEXLang (we’ve been waiting long enough), those giant SQL export files turn into structured files you can actually parse. Suddenly the snake can hunt page processes calling tables directly and all the other crimes hiding in exports.

Same engine.
Just better food.
Let the snake hiss.