Warning

You are viewing the documentation for the development version. The latest stable release can be found at kilo59.github.io/ruff-sync.

Index

ruff_sync

Ruff sync package.

This package provides tools to synchronize ruff configuration across projects.

__version__ module-attribute

__version__ = version('ruff-sync')

get_ruff_tool_table module-attribute

get_ruff_tool_table = get_ruff_config

__all__ module-attribute

__all__ = [
    "Arguments",
    "Config",
    "FetchResult",
    "OutputFormat",
    "RuffConfigFileName",
    "__version__",
    "check",
    "fetch_upstream_config",
    "get_config",
    "get_formatter",
    "get_ruff_config",
    "get_ruff_tool_table",
    "inspect",
    "is_ruff_toml_file",
    "load_local_ruff_config",
    "main",
    "merge_ruff_toml",
    "pull",
    "resolve_raw_url",
    "resolve_target_path",
    "to_git_url",
    "toml_ruff_parse",
]

Arguments

Bases: NamedTuple

CLI arguments for the ruff-sync tool.

Source code in src/ruff_sync/cli.py

class Arguments(NamedTuple):
    """CLI arguments for the ruff-sync tool."""

    command: str
    upstream: tuple[URL, ...]
    to: pathlib.Path
    exclude: Iterable[str]
    verbose: int
    branch: str = DEFAULT_BRANCH
    path: str | None = None
    semantic: bool = False
    diff: bool = True
    init: bool = False
    pre_commit: bool = False
    save: bool | None = None
    output_format: OutputFormat = OutputFormat.TEXT

    @property
    @deprecated("Use 'to' instead")
    def source(self) -> pathlib.Path:
        """Deprecated: use 'to' instead."""
        return self.to

    @classmethod
    @lru_cache(maxsize=1)
    def fields(cls) -> set[str]:
        """Return the set of all field names, including deprecated ones."""
        return set(cls._fields) | {"source"}

command instance-attribute

command

upstream instance-attribute

upstream

to instance-attribute

to

exclude instance-attribute

exclude

verbose instance-attribute

verbose

branch class-attribute instance-attribute

branch = DEFAULT_BRANCH

path class-attribute instance-attribute

path = None

semantic class-attribute instance-attribute

semantic = False

diff class-attribute instance-attribute

diff = True

init class-attribute instance-attribute

init = False

pre_commit class-attribute instance-attribute

pre_commit = False

save class-attribute instance-attribute

save = None

output_format class-attribute instance-attribute

output_format = TEXT

source property

source

Deprecated: use 'to' instead.

fields cached classmethod

fields()

Return the set of all field names, including deprecated ones.

Source code in src/ruff_sync/cli.py

@classmethod
@lru_cache(maxsize=1)
def fields(cls) -> set[str]:
    """Return the set of all field names, including deprecated ones."""
    return set(cls._fields) | {"source"}

RuffConfigFileName

Bases: str, Enum

Enumeration of Ruff configuration filenames.

Source code in src/ruff_sync/config_io.py

@enum.unique
class RuffConfigFileName(str, enum.Enum):
    """Enumeration of Ruff configuration filenames."""

    PYPROJECT_TOML = "pyproject.toml"
    RUFF_TOML = "ruff.toml"
    DOT_RUFF_TOML = ".ruff.toml"

    @classmethod
    def tried_order(cls) -> list[RuffConfigFileName]:
        """Return the order in which configuration files should be tried."""
        return [cls.RUFF_TOML, cls.DOT_RUFF_TOML, cls.PYPROJECT_TOML]

    @override
    def __str__(self) -> str:
        """Return the filename as a string."""
        return self.value

PYPROJECT_TOML class-attribute instance-attribute

PYPROJECT_TOML = 'pyproject.toml'

RUFF_TOML class-attribute instance-attribute

RUFF_TOML = 'ruff.toml'

DOT_RUFF_TOML class-attribute instance-attribute

DOT_RUFF_TOML = '.ruff.toml'

tried_order classmethod

tried_order()

Return the order in which configuration files should be tried.

Source code in src/ruff_sync/config_io.py

@classmethod
def tried_order(cls) -> list[RuffConfigFileName]:
    """Return the order in which configuration files should be tried."""
    return [cls.RUFF_TOML, cls.DOT_RUFF_TOML, cls.PYPROJECT_TOML]

__str__

__str__()

Return the filename as a string.

Source code in src/ruff_sync/config_io.py

@override
def __str__(self) -> str:
    """Return the filename as a string."""
    return self.value

OutputFormat

Bases: str, Enum

Output formats for the CLI.

Source code in src/ruff_sync/constants.py

@enum.unique
class OutputFormat(str, enum.Enum):
    """Output formats for the CLI."""

    TEXT = "text"
    JSON = "json"
    GITHUB = "github"
    GITLAB = "gitlab"
    SARIF = "sarif"

    @override
    def __str__(self) -> str:
        """Return the string value for argparse help."""
        return self.value

TEXT class-attribute instance-attribute

TEXT = 'text'

JSON class-attribute instance-attribute

JSON = 'json'

GITHUB class-attribute instance-attribute

GITHUB = 'github'

GITLAB class-attribute instance-attribute

GITLAB = 'gitlab'

SARIF class-attribute instance-attribute

SARIF = 'sarif'

__str__

__str__()

Return the string value for argparse help.

Source code in src/ruff_sync/constants.py

@override
def __str__(self) -> str:
    """Return the string value for argparse help."""
    return self.value

Config

Bases: TypedDict

Configuration schema for [tool.ruff-sync] in pyproject.toml.

Source code in src/ruff_sync/core.py

class Config(TypedDict, total=False):
    """Configuration schema for [tool.ruff-sync] in pyproject.toml."""

    upstream: str | list[str]
    to: str
    source: str  # Deprecated
    exclude: list[str]
    verbose: int
    branch: str
    path: str
    semantic: bool
    diff: bool
    init: bool
    pre_commit_version_sync: bool
    output_format: str

upstream instance-attribute

upstream

to instance-attribute

to

source instance-attribute

source

exclude instance-attribute

exclude

verbose instance-attribute

verbose

branch instance-attribute

branch

path instance-attribute

path

semantic instance-attribute

semantic

diff instance-attribute

diff

init instance-attribute

init

pre_commit_version_sync instance-attribute

pre_commit_version_sync

output_format instance-attribute

output_format

FetchResult

Bases: NamedTuple

Result of fetching an upstream configuration.

Source code in src/ruff_sync/core.py

class FetchResult(NamedTuple):
    """Result of fetching an upstream configuration."""

    buffer: StringIO
    resolved_upstream: URL

buffer instance-attribute

buffer

resolved_upstream instance-attribute

resolved_upstream

get_config cached

get_config(source)

Read [tool.ruff-sync] configuration from pyproject.toml.

Examples:

>>> import pathlib
>>> config = get_config(pathlib.Path("."))
>>> if "upstream" in config:
...     print(f"Syncing from {config['upstream']}")

Source code in src/ruff_sync/cli.py

@lru_cache(maxsize=1)
def get_config(
    source: pathlib.Path,
) -> Config:
    """Read [tool.ruff-sync] configuration from pyproject.toml.

    Examples:
        >>> import pathlib
        >>> config = get_config(pathlib.Path("."))
        >>> if "upstream" in config:
        ...     print(f"Syncing from {config['upstream']}")
    """
    local_toml = source / RuffConfigFileName.PYPROJECT_TOML
    # TODO: use pydantic to validate the toml file
    cfg_result: Config = {}
    if local_toml.exists():
        toml = tomlkit.parse(local_toml.read_text())
        config = toml.get("tool", {}).get("ruff-sync")
        if config:
            allowed_keys = set(Config.__annotations__.keys())
            for raw_key, value in config.items():
                # Check for legacy 'source' key to emit deprecation warning
                if raw_key.replace("-", "_") == ConfKey.to_attr(ConfKey.SOURCE):
                    LOGGER.warning(
                        f"DeprecationWarning: [tool.ruff-sync] '{raw_key}' "
                        f"is deprecated. Use '{ConfKey.TO}' instead."
                    )

                # Map legacy names (source, pre_commit_sync) to canonical
                # (to, pre-commit-version-sync)
                canonical_key = ConfKey.get_canonical(raw_key)

                # Normalize TOML key (dashes) to internal Python attribute name (underscores)
                # e.g. "pre-commit-version-sync" -> "pre_commit_version_sync"
                arg_attr = ConfKey.to_attr(canonical_key)

                if arg_attr in allowed_keys:
                    cfg_result[arg_attr] = value  # type: ignore[literal-required]
                else:
                    LOGGER.warning(f"Unknown ruff-sync configuration: {raw_key}")

            # Ensure 'to' is populated if 'source' was used
            to_attr = ConfKey.to_attr(ConfKey.TO)
            source_attr = ConfKey.to_attr(ConfKey.SOURCE)
            if source_attr in cfg_result and to_attr not in cfg_result:
                cfg_result[to_attr] = cfg_result[source_attr]  # type: ignore[literal-required]
    return cfg_result

inspect

inspect()

Entry point for the ruff-inspect console script.

Source code in src/ruff_sync/cli.py

def inspect() -> int:
    """Entry point for the ruff-inspect console script."""
    # Handle optional subcommands/args if user passed any to ruff-inspect
    # but primarily ensure "inspect" is the command.
    if len(sys.argv) > 1 and sys.argv[1] not in ("-h", "--help", "--version"):
        # If they passed args but no command, insert 'inspect'
        if sys.argv[1] not in ("pull", "check", "inspect"):
            sys.argv.insert(1, "inspect")
    else:
        # Default to 'inspect' if no args or just flags
        sys.argv.insert(1, "inspect")

    return main()

main

main()

Run the ruff-sync CLI.

Source code in src/ruff_sync/cli.py

def main() -> int:
    """Run the ruff-sync CLI."""
    # Handle backward compatibility: default to 'pull' if no command provided
    if len(sys.argv) > 1 and sys.argv[1] not in (
        "pull",
        "check",
        "inspect",
        "-h",
        "--help",
        "--version",
    ):
        sys.argv.insert(1, "pull")
    elif len(sys.argv) == 1:
        sys.argv.append("pull")

    args = PARSER.parse_args()

    # Configure logging
    log_level = {
        0: logging.WARNING,
        1: logging.INFO,
    }.get(args.verbose, logging.DEBUG)

    # Configure logging for the entire ruff_sync package
    root_logger = logging.getLogger("ruff_sync")
    root_logger.setLevel(log_level)

    # Avoid adding multiple handlers if main() is called multiple times (e.g. in tests)
    if not root_logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(ColoredFormatter())
        root_logger.addHandler(handler)

    root_logger.propagate = "PYTEST_CURRENT_TEST" in os.environ

    # Determine target 'to' from CLI or use default '.'
    # Defer Path conversion to avoid pyfakefs issues with captured Path class
    arg_to = args.to or args.source
    initial_to = pathlib.Path(arg_to) if arg_to else pathlib.Path()
    config: Config = get_config(initial_to)

    upstream, to_val, exclude, branch, path, output_format = _resolve_args(args, config, initial_to)
    pre_commit_val = _resolve_pre_commit(args, config)

    resolved_upstream = tuple(resolve_raw_url(u, branch=branch, path=path) for u in upstream)

    exec_args = Arguments(
        command=args.command,
        upstream=resolved_upstream,
        to=to_val,
        exclude=exclude,
        verbose=args.verbose,
        branch=branch,
        path=path,
        semantic=getattr(args, "semantic", False),
        diff=getattr(args, "diff", True),
        init=getattr(args, "init", False),
        pre_commit=pre_commit_val,
        save=getattr(args, "save", None),
        output_format=output_format,
    )

    # Warn if the specified output format doesn't match the current CI environment
    _validate_ci_output_format(exec_args)

    try:
        if exec_args.command == "inspect":
            from ruff_sync.tui import get_tui_app  # noqa: PLC0415

            app_class = get_tui_app()
            return app_class(exec_args).run() or 0

        if exec_args.command == "check":
            return asyncio.run(check(exec_args))
        return asyncio.run(pull(exec_args))
    except DependencyError as e:
        LOGGER.error(f"❌ {e}")  # noqa: TRY400
        return 1
    except UpstreamError as e:
        for url, err in e.errors:
            LOGGER.error(f"❌ Failed to fetch {url}: {err}")  # noqa: TRY400
        return 4

is_ruff_toml_file

is_ruff_toml_file(path_or_url)

Return True if the path or URL indicates a ruff.toml file.

This handles cases with query strings or fragments by examining only the path component.

Source code in src/ruff_sync/config_io.py

def is_ruff_toml_file(path_or_url: str | URL) -> bool:
    """Return True if the path or URL indicates a ruff.toml file.

    This handles cases with query strings or fragments by examining only the path component.
    """
    parsed = urlparse(str(path_or_url))

    # If it's a URL with a scheme/netloc, use the parsed path component.
    # Otherwise, fall back to stripping any query/fragment from the raw string.
    if parsed.scheme or parsed.netloc:
        path = parsed.path
    else:
        path = str(path_or_url).split("?", 1)[0].split("#", 1)[0]

    return pathlib.Path(path).name in (
        RuffConfigFileName.RUFF_TOML,
        RuffConfigFileName.DOT_RUFF_TOML,
    )

load_local_ruff_config

load_local_ruff_config(target)

Load the local Ruff configuration as a plain dictionary.

Parameters:

Name	Type	Description	Default
target	Path	The directory or file path to load configuration from.	required

Returns:

Type	Description
dict[str, Any]	A plain dictionary containing the [tool.ruff] configuration.

Raises:

Type	Description
FileNotFoundError	If no configuration file is found at the target path.
TypeError	If the configuration structure is invalid.

Source code in src/ruff_sync/config_io.py

def load_local_ruff_config(target: pathlib.Path) -> dict[str, Any]:
    """Load the local Ruff configuration as a plain dictionary.

    Args:
        target: The directory or file path to load configuration from.

    Returns:
        A plain dictionary containing the [tool.ruff] configuration.

    Raises:
        FileNotFoundError: If no configuration file is found at the target path.
        TypeError: If the configuration structure is invalid.
    """
    config_path = resolve_target_path(target)
    if not config_path.exists():
        msg = f"No Ruff configuration file found at: {config_path}"
        raise FileNotFoundError(msg)

    content = config_path.read_text(encoding="utf-8")
    doc = tomlkit.parse(content)

    is_ruff_toml = config_path.name in (
        RuffConfigFileName.RUFF_TOML,
        RuffConfigFileName.DOT_RUFF_TOML,
    )

    if is_ruff_toml:
        # For ruff.toml, the entire document is the configuration
        unwrapped = doc.unwrap() if hasattr(doc, "unwrap") else doc
        return unwrapped if isinstance(unwrapped, dict) else {}

    # For pyproject.toml, extract [tool.ruff]
    try:
        # We use isinstance checks instead of cast to satisfy mypy
        tool = doc.get("tool")
        if not isinstance(tool, Mapping):
            return {}
        ruff = tool.get("ruff")
        if ruff is None:
            return {}
        if not isinstance(ruff, (Mapping, Table)):
            msg = f"Expected table for [tool.ruff], got {type(ruff)}"
            raise TypeError(msg)

        data = ruff.unwrap() if hasattr(ruff, "unwrap") else ruff
        if not isinstance(data, dict):
            return {}
    except KeyError:
        return {}
    else:
        return data

resolve_target_path

resolve_target_path(to, upstreams=None)

Resolve the target path for configuration files.

If 'to' is a file, it's used directly. Otherwise, it looks for existing ruff/pyproject.toml in the 'to' directory. If none found, it defaults to pyproject.toml unless the first upstream is a ruff.toml.

Source code in src/ruff_sync/config_io.py

def resolve_target_path(
    to: pathlib.Path, upstreams: Iterable[str | URL] | None = None
) -> pathlib.Path:
    """Resolve the target path for configuration files.

    If 'to' is a file, it's used directly.
    Otherwise, it looks for existing ruff/pyproject.toml in the 'to' directory.
    If none found, it defaults to pyproject.toml unless the first upstream is a ruff.toml.
    """
    if to.is_file():
        return to

    # If it's a directory, look for common config files
    for filename in RuffConfigFileName.tried_order():
        candidate = to / filename
        if candidate.exists():
            return candidate

    # Use the first upstream URL as a hint for the default file name
    first_upstream = next(iter(upstreams), None) if upstreams else None

    # If upstream is specified and is a ruff.toml, default to ruff.toml
    if first_upstream and is_ruff_toml_file(first_upstream):
        return to / RuffConfigFileName.RUFF_TOML

    return to / RuffConfigFileName.PYPROJECT_TOML

check async

check(args)

Check if the local pyproject.toml / ruff.toml is in sync with the upstream.

Returns:

Name	Type	Description
int	int	0 if in sync, 1 if out of sync.

Examples:

>>> import asyncio
>>> from ruff_sync.cli import Arguments
>>> from httpx import URL
>>> import pathlib
>>> args = Arguments(
...     command="check",
...     upstream=URL("https://github.com/org/repo/blob/main/pyproject.toml"),
...     to=pathlib.Path("pyproject.toml"),
...     exclude=[],
... )
>>> # asyncio.run(check(args))

Source code in src/ruff_sync/core.py

async def check(
    args: Arguments,
) -> int:
    """Check if the local pyproject.toml / ruff.toml is in sync with the upstream.

    Returns:
        int: 0 if in sync, 1 if out of sync.

    Examples:
        >>> import asyncio
        >>> from ruff_sync.cli import Arguments
        >>> from httpx import URL
        >>> import pathlib
        >>> args = Arguments(
        ...     command="check",
        ...     upstream=URL("https://github.com/org/repo/blob/main/pyproject.toml"),
        ...     to=pathlib.Path("pyproject.toml"),
        ...     exclude=[],
        ... )
        >>> # asyncio.run(check(args))
    """
    output_format = args.output_format
    fmt: ResultFormatter = get_formatter(output_format)
    try:
        fmt.note("🔍 Checking Ruff sync status...")

        _source_toml_path = resolve_target_path(args.to, args.upstream).resolve(strict=False)
        if not _source_toml_path.exists():
            fmt.error(
                f"❌ Configuration file {_source_toml_path} does not exist. "
                "Run 'ruff-sync pull' to create it.",
                file_path=_source_toml_path,
                logger=LOGGER,
            )
            return 1

        source_toml_file = TOMLFile(_source_toml_path)
        source_doc = source_toml_file.read()

        # Create a copy for comparison
        source_doc_copy = tomlkit.parse(source_doc.as_string())
        merged_doc = source_doc_copy

        async with httpx.AsyncClient() as client:
            merged_doc = await _merge_multiple_upstreams(
                merged_doc,
                is_target_ruff_toml=is_ruff_toml_file(_source_toml_path.name),
                args=args,
                client=client,
            )

        is_source_ruff_toml = is_ruff_toml_file(_source_toml_path.name)
        source_val: object = None
        merged_val: object = None
        if args.semantic:
            if is_source_ruff_toml:
                source_ruff = source_doc
                merged_ruff = merged_doc
            else:
                source_ruff = source_doc.get("tool", {}).get("ruff")
                merged_ruff = merged_doc.get("tool", {}).get("ruff")

            # Compare unwrapped versions
            source_val = source_ruff.unwrap() if source_ruff is not None else None
            merged_val = merged_ruff.unwrap() if merged_ruff is not None else None

            if source_val == merged_val:
                fmt.success("✅ Ruff configuration is semantically in sync.")
                exit_code = _check_pre_commit_sync(args, fmt)
                if exit_code is not None:
                    return exit_code
                return 0
        elif source_doc.as_string() == merged_doc.as_string():
            fmt.success("✅ Ruff configuration is in sync.")
            exit_code = _check_pre_commit_sync(args, fmt)
            if exit_code is not None:
                return exit_code
            return 0

        try:
            rel_path = _source_toml_path.relative_to(pathlib.Path.cwd())
        except ValueError:
            rel_path = _source_toml_path

        _report_drift(
            fmt=fmt,
            rel_path=rel_path,
            source_doc=source_doc,
            merged_doc=merged_doc,
            is_ruff_toml=is_ruff_toml_file(_source_toml_path.name),
        )

        if args.diff:
            _print_diff(
                args=args,
                fmt=fmt,
                ctx=DiffContext(
                    source_toml_path=_source_toml_path,
                    source_doc=source_doc,
                    merged_doc=merged_doc,
                    source_val=source_val,
                    merged_val=merged_val,
                ),
            )
        return 1
    finally:
        fmt.finalize()

fetch_upstream_config async

fetch_upstream_config(url, client, branch, path)

Fetch the upstream pyproject.toml either via HTTP or git clone.

Source code in src/ruff_sync/core.py

async def fetch_upstream_config(
    url: URL, client: httpx.AsyncClient, branch: str, path: str | None
) -> FetchResult:
    """Fetch the upstream pyproject.toml either via HTTP or git clone."""
    if is_git_url(url):
        LOGGER.info(f"Cloning {url} via git...")
        return await asyncio.to_thread(_fetch_via_git, url, branch, path)

    try:
        return await _download_with_discovery(url, client, branch)
    except httpx.HTTPStatusError as err:
        msg = f"HTTP error {err.response.status_code} when downloading from {url}"
        git_url = to_git_url(url)
        if git_url:
            # sys.argv[1] might be -v or something else when running via pytest
            try:
                cmd = sys.argv[1]
                if cmd not in ("pull", "check"):
                    cmd = "pull"
            except IndexError:
                cmd = "pull"
            msg += (
                f"\n\n💡 Check the URL and your permissions. "
                "You might want to try cloning via git instead:\n\n"
                f"   ruff-sync {cmd} {git_url}"
            )
        else:
            msg += "\n\n💡 Check the URL and your permissions."

        # Re-raise with a more helpful message while preserving the original exception context
        raise httpx.HTTPStatusError(msg, request=err.request, response=err.response) from None

get_ruff_config

get_ruff_config(
    toml: str | TOMLDocument,
    is_ruff_toml: bool = ...,
    create_if_missing: Literal[True] = ...,
    exclude: Iterable[str] = ...,
) -> TOMLDocument | Table

get_ruff_config(
    toml: str | TOMLDocument,
    is_ruff_toml: bool = ...,
    create_if_missing: Literal[False] = ...,
    exclude: Iterable[str] = ...,
) -> TOMLDocument | Table | None

get_ruff_config(
    toml,
    is_ruff_toml=False,
    create_if_missing=True,
    exclude=(),
)

Get the ruff section or document from a TOML string.

If it does not exist and it is a pyproject.toml, create it.

Source code in src/ruff_sync/core.py

def get_ruff_config(
    toml: str | TOMLDocument,
    is_ruff_toml: bool = False,
    create_if_missing: bool = True,
    exclude: Iterable[str] = (),
) -> TOMLDocument | Table | None:
    """Get the ruff section or document from a TOML string.

    If it does not exist and it is a pyproject.toml, create it.
    """
    if isinstance(toml, str):
        doc: TOMLDocument = tomlkit.parse(toml)
    else:
        doc = toml

    if is_ruff_toml:
        _apply_exclusions(doc, exclude)
        return doc

    try:
        tool: Table = doc["tool"]  # type: ignore[assignment]
        ruff = tool["ruff"]
        LOGGER.debug("Found `tool.ruff` section.")
    except KeyError:
        if not create_if_missing:
            return None
        LOGGER.info("✨ No `tool.ruff` section found, creating it.")
        tool = table(True)
        ruff = table()
        tool.append("ruff", ruff)
        doc.append("tool", tool)
    if not isinstance(ruff, Table):
        msg = f"Expected table, got {type(ruff)}"
        raise TypeError(msg)
    _apply_exclusions(ruff, exclude)
    return ruff

merge_ruff_toml

merge_ruff_toml(
    source, upstream_ruff_doc, is_ruff_toml=False
)

Merge the source and upstream tool ruff config with better whitespace preservation.

Examples:

>>> from tomlkit import parse
>>> source = parse("[tool.ruff]\nline-length = 80")
>>> upstream = parse("[tool.ruff]\nline-length = 100")["tool"]["ruff"]
>>> merged = merge_ruff_toml(source, upstream)
>>> print(merged.as_string())
[tool.ruff]
line-length = 100

Source code in src/ruff_sync/core.py

def merge_ruff_toml(
    source: TOMLDocument,
    upstream_ruff_doc: TOMLDocument | Table | None,
    is_ruff_toml: bool = False,
) -> TOMLDocument:
    r"""Merge the source and upstream tool ruff config with better whitespace preservation.

    Examples:
        >>> from tomlkit import parse
        >>> source = parse("[tool.ruff]\nline-length = 80")
        >>> upstream = parse("[tool.ruff]\nline-length = 100")["tool"]["ruff"]
        >>> merged = merge_ruff_toml(source, upstream)
        >>> print(merged.as_string())
        [tool.ruff]
        line-length = 100
    """
    if not upstream_ruff_doc:
        LOGGER.warning("No upstream ruff config section found.")
        return source

    if is_ruff_toml:
        _recursive_update(source, upstream_ruff_doc)
        return source

    source_tool_ruff = get_ruff_config(source, create_if_missing=True)

    _recursive_update(source_tool_ruff, upstream_ruff_doc)

    # Add a blank separator line after the ruff section — but only when another
    # top-level section follows it. Adding \n\n at end-of-file is unnecessary.
    doc_str = source.as_string()
    ruff_start = doc_str.find("[tool.ruff]")
    # Look for any non-ruff top-level section header after [tool.ruff]
    ruff_is_last = ruff_start == -1 or not re.search(
        r"^\[(?!tool\.ruff)", doc_str[ruff_start:], re.MULTILINE
    )
    if not ruff_is_last and not source_tool_ruff.as_string().endswith("\n\n"):
        source_tool_ruff.add(tomlkit.nl())

    return source

pull async

pull(args)

Pull the upstream ruff config and apply it to the source.

Returns:

Name	Type	Description
int	int	0 on success, 1 on failure.

Examples:

>>> import asyncio
>>> from ruff_sync.cli import Arguments
>>> from httpx import URL
>>> import pathlib
>>> args = Arguments(
...     command="pull",
...     upstream=URL("https://github.com/org/repo/blob/main/pyproject.toml"),
...     to=pathlib.Path("pyproject.toml"),
...     exclude=["lint.isort"],
...     init=True,
... )
>>> # asyncio.run(pull(args))

Source code in src/ruff_sync/core.py

async def pull(
    args: Arguments,
) -> int:
    """Pull the upstream ruff config and apply it to the source.

    Returns:
        int: 0 on success, 1 on failure.

    Examples:
        >>> import asyncio
        >>> from ruff_sync.cli import Arguments
        >>> from httpx import URL
        >>> import pathlib
        >>> args = Arguments(
        ...     command="pull",
        ...     upstream=URL("https://github.com/org/repo/blob/main/pyproject.toml"),
        ...     to=pathlib.Path("pyproject.toml"),
        ...     exclude=["lint.isort"],
        ...     init=True,
        ... )
        >>> # asyncio.run(pull(args))
    """
    output_format = args.output_format
    fmt = get_formatter(output_format)
    try:
        fmt.note("🔄 Syncing Ruff...")
        _source_toml_path = resolve_target_path(args.to, args.upstream).resolve(strict=False)

        source_toml_file = TOMLFile(_source_toml_path)
        if _source_toml_path.exists():
            source_doc = source_toml_file.read()
        elif args.init:
            LOGGER.info(f"✨ Target file {_source_toml_path} does not exist, creating it.")
            source_doc = tomlkit.document()
            # Scaffold the file immediately to ensure we can write to the enclosing directory
            try:
                _source_toml_path.parent.mkdir(parents=True, exist_ok=True)
                _source_toml_path.touch()
            except OSError as e:
                fmt.error(f"❌ Failed to create {_source_toml_path}: {e}", logger=LOGGER)
                return 1
        else:
            fmt.error(
                f"❌ Configuration file {_source_toml_path} does not exist. "
                "Pass the '--init' flag to create it.",
                file_path=_source_toml_path,
                logger=LOGGER,
            )
            return 1

        async with httpx.AsyncClient() as client:
            source_doc = await _merge_multiple_upstreams(
                source_doc,
                is_target_ruff_toml=is_ruff_toml_file(_source_toml_path.name),
                args=args,
                client=client,
            )

        should_save = args.save if args.save is not None else args.init
        if should_save:
            if _source_toml_path.name == RuffConfigFileName.PYPROJECT_TOML:
                LOGGER.info(f"Saving [tool.ruff-sync] configuration to {_source_toml_path.name}")
                serialize_ruff_sync_config(source_doc, args)
            else:
                LOGGER.info(
                    "Skipping [tool.ruff-sync] configuration save "
                    "because target is not pyproject.toml"
                )

        source_toml_file.write(source_doc)
        try:
            rel_path = _source_toml_path.resolve().relative_to(pathlib.Path.cwd())
        except ValueError:
            rel_path = _source_toml_path.resolve()
        fmt.success(f"✅ Updated {rel_path}")

        if args.pre_commit:
            sync_pre_commit(pathlib.Path.cwd(), dry_run=False)

        return 0
    finally:
        fmt.finalize()

resolve_raw_url

resolve_raw_url(url, branch='main', path=None)

Convert a GitHub or GitLab repository/blob URL to a raw content URL.

Parameters:

Name	Type	Description	Default
url	URL	The URL to resolve.	required
branch	str	The default branch to use for repo URLs.	'main'
path	str | None	The directory prefix for pyproject.toml.	None

Returns:

Name	Type	Description
URL	URL	The resolved raw content URL, or the original URL if no conversion applies.

Source code in src/ruff_sync/core.py

def resolve_raw_url(url: URL, branch: str = "main", path: str | None = None) -> URL:
    """Convert a GitHub or GitLab repository/blob URL to a raw content URL.

    Args:
        url (URL): The URL to resolve.
        branch (str): The default branch to use for repo URLs.
        path (str | None): The directory prefix for pyproject.toml.

    Returns:
        URL: The resolved raw content URL, or the original URL if no conversion applies.

    """
    # If it's a git URL, leave it alone; we'll handle it via git clone
    if is_git_url(url):
        return url
    LOGGER.debug(f"Initial URL: {url}")
    if url.host in _GITHUB_HOSTS:
        return _convert_github_url(url, branch=branch, path=path or "")
    if url.host in _GITLAB_HOSTS:
        return _convert_gitlab_url(url, branch=branch, path=path or "")
    return url

to_git_url

to_git_url(url)

Attempt to convert a browser or raw URL to a git (SSH) URL.

Supports GitHub and GitLab.

Source code in src/ruff_sync/core.py

def to_git_url(url: URL) -> URL | None:
    """Attempt to convert a browser or raw URL to a git (SSH) URL.

    Supports GitHub and GitLab.
    """
    if is_git_url(url):
        return url

    if url.host in _GITHUB_HOSTS or url.host == _GITHUB_RAW_HOST:
        path_parts = [p for p in url.path.split("/") if p]
        if len(path_parts) >= _GITHUB_REPO_PATH_PARTS_COUNT:
            org, repo = path_parts[:_GITHUB_REPO_PATH_PARTS_COUNT]
            repo = repo.removesuffix(".git")
            return URL(f"git@github.com:{org}/{repo}.git")

    if url.host in _GITLAB_HOSTS:
        path = url.path.strip("/")
        project_path = path.split("/-/")[0] if "/-/" in path else path
        if project_path:
            project_path = project_path.removesuffix(".git")
            return URL(f"git@{url.host}:{project_path}.git")

    return None

toml_ruff_parse

toml_ruff_parse(toml_s, exclude)

Parse a TOML string for the tool.ruff section excluding certain ruff configs.

Source code in src/ruff_sync/core.py

def toml_ruff_parse(toml_s: str, exclude: Iterable[str]) -> TOMLDocument:
    """Parse a TOML string for the tool.ruff section excluding certain ruff configs."""
    ruff_toml: TOMLDocument = tomlkit.parse(toml_s)["tool"]["ruff"]  # type: ignore[index,assignment]
    for section in exclude:
        LOGGER.info(f"Excluding section `lint.{section}` from ruff config.")
        ruff_toml["lint"].pop(section, None)  # type: ignore[union-attr]
    return ruff_toml

get_formatter

get_formatter(output_format)

Return the corresponding formatter for the given format.

Source code in src/ruff_sync/formatters.py

def get_formatter(output_format: OutputFormat) -> ResultFormatter:
    """Return the corresponding formatter for the given format."""
    match output_format:
        case OutputFormat.TEXT:
            return TextFormatter()
        case OutputFormat.GITHUB:
            return GithubFormatter()
        case OutputFormat.JSON:
            return JsonFormatter()
        case OutputFormat.GITLAB:
            return GitlabFormatter()
        case OutputFormat.SARIF:
            return SarifFormatter()