Core Module Reference

The core module provides the fundamental workflow and transformation functions.

Workflows

core.workflow

Define a code synchronization workflow:

core.workflow(
    name = "string",                    # Required: unique identifier
    origin = origin,                    # Required: source of code
    destination = destination,          # Required: target for code
    authoring = authoring,              # Required: author handling
    origin_files = glob(["**"]),        # Optional: files to read
    destination_files = glob(["**"]),   # Optional: files to manage
    transformations = [],               # Optional: transformations
    mode = "SQUASH",                    # Optional: commit mode
    reversible_check = False,           # Optional: verify reversibility
    check_last_rev_state = True,        # Optional: validate state
    ask_for_confirmation = False,       # Optional: interactive confirm
)

core.feedback

Define an event reaction workflow:

core.feedback(
    name = "string",                    # Required: unique identifier
    origin = trigger,                   # Required: event source
    destination = api,                  # Required: API destination
    actions = [],                       # Required: actions to perform
)

Transformations

core.move

Move files or directories:

core.move(
    before = "source/path/",            # Source path
    after = "dest/path/",               # Destination path
    paths = glob(["**"]),               # Optional: filter
    overwrite = False,                  # Optional: overwrite existing
)

core.copy

Copy files (keep originals):

core.copy(
    before = "source",                  # Source path
    after = "destination",              # Destination path
    paths = glob(["**"]),               # Optional: filter
    overwrite = False,                  # Optional: overwrite existing
)

core.remove

Remove files:

core.remove(
    glob(["**/internal/**"]),           # Files to remove
)

core.rename

Rename files (simplified version of core.move for renaming):

core.rename(
    before = ".internal.js",            # Filepath or suffix to change
    after = ".js",                      # Replacement filepath or suffix
    paths = glob(["**"]),               # Optional: file filter
    overwrite = False,                  # Optional: overwrite existing
)

Note: core.rename only renames regular files, not directories.

core.todo_replace

Replace Google-style TODOs like TODO(username) or NOTE(user1, user2):

core.todo_replace(
    tags = ["TODO", "NOTE"],            # Tags to look for (default)
    mapping = {"user1": "newuser1"},    # Username mapping
    mode = "MAP_OR_IGNORE",             # How to handle unmapped users
    paths = glob(["**"]),               # Optional: file filter
    default = None,                     # Default value for MAP_OR_DEFAULT/USE_DEFAULT
    ignore = None,                      # Regex pattern to ignore
)

Modes:

Mode	Description
`MAP_OR_FAIL`	Use mapping, fail if user not found
`MAP_OR_IGNORE`	Use mapping, keep original if not found (default)
`MAP_OR_DEFAULT`	Use mapping, use `default` if not found
`SCRUB_NAMES`	Remove all names: `TODO(foo)` → `TODO`
`USE_DEFAULT`	Replace all with default: `TODO(foo)` → `TODO(default)`

Example - Scrub names for open sourcing:

core.todo_replace(
    mode = "SCRUB_NAMES",
)
# TODO(alice, bob): Fix this → TODO: Fix this

Example - Map to external usernames:

core.todo_replace(
    mapping = {
        "alice": "alice-github",
        "bob": "bob-github",
    },
    mode = "MAP_OR_DEFAULT",
    default = "nicetomeetyou",
)
# TODO(alice): Fix → TODO(alice-github): Fix
# TODO(unknown): Fix → TODO(nicetomeetyou): Fix

Example - Ignore certain patterns:

core.todo_replace(
    mapping = {"alice": "external"},
    ignore = "b/[0-9]+",  # Ignore bug references like b/123456
)
# TODO(b/123456, alice): Fix → TODO(b/123456, external): Fix

core.replace

Text replacement:

core.replace(
    before = "old_text",                # Text to find
    after = "new_text",                 # Replacement text
    regex_groups = {},                  # Optional: regex capture groups
    paths = glob(["**"]),               # Optional: file filter
    first_only = False,                 # Optional: first match only
    multiline = False,                  # Optional: multiline mode
)

core.filter_replace

Replacement with mapping:

core.filter_replace(
    regex = "pattern",                  # Pattern to match
    mapping = {"key": "value"},         # Replacement mapping
    default = None,                     # Optional: default if no match
    paths = glob(["**"]),               # Optional: file filter
)

core.verify_match

Verify pattern presence:

core.verify_match(
    regex = "pattern",                  # Pattern to check
    paths = glob(["**"]),               # Optional: file filter
    verify_no_match = False,            # If True, fail if pattern found
)

core.transform

Group transformations with reversal:

core.transform(
    transformations = [],               # Forward transformations
    reversal = [],                      # Reverse transformations
    ignore_noop = False,                # Don't fail on no-op
)

core.dynamic_transform

Custom Starlark transformation:

def _my_transform(ctx):
    for f in ctx.run(glob(["**/*.txt"])):
        content = ctx.read_path(f)
        ctx.write_path(f, content.upper())
    return ctx.success()

core.dynamic_transform(impl = _my_transform)

Actions

core.action

Define a feedback action:

def _my_action(ctx):
    ctx.destination.create_issue(title = "...", body = "...")
    return ctx.success()

core.action(
    impl = _my_action,
    params = {},
)

Glob Function

glob

Create file patterns:

# Include patterns
glob(["src/**", "docs/**"])

# Include and exclude
glob(
    include = ["**"],
    exclude = ["**/test/**"],
)