Warning

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

core

ruff_sync.core

Core logic for ruff-sync.

__all__ module-attribute

__all__ = [
    "Config",
    "FetchResult",
    "RuffConfigFileName",
    "UpstreamError",
    "check",
    "fetch_upstream_config",
    "fetch_upstreams_concurrently",
    "get_ruff_config",
    "get_ruff_tool_table",
    "is_ruff_toml_file",
    "merge_ruff_toml",
    "pull",
    "resolve_raw_url",
    "resolve_target_path",
    "serialize_ruff_sync_config",
    "to_git_url",
    "toml_ruff_parse",
]

LOGGER module-attribute

LOGGER = getLogger(__name__)

get_ruff_tool_table module-attribute

get_ruff_tool_table = get_ruff_config

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

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

UpstreamError

Bases: Exception

Raised when one or more upstream fetches fail.

Attributes:

Name	Type	Description
errors	Final[tuple[tuple[URL, BaseException], ...]]	A tuple of tuples containing the URL and the BaseException that occurred.

Source code in src/ruff_sync/core.py

class UpstreamError(Exception):
    """Raised when one or more upstream fetches fail.

    Attributes:
        errors: A tuple of tuples containing the URL and the BaseException that occurred.
    """

    def __init__(self, errors: Iterable[tuple[URL, BaseException]]) -> None:
        """Initialize UpstreamError with a list of fetch failures."""
        self.errors: Final[tuple[tuple[URL, BaseException], ...]] = tuple(errors)
        error_count = len(self.errors)
        msg = f"❌ {error_count} upstream fetch{'es' if error_count > 1 else ''} failed"
        super().__init__(msg)

errors instance-attribute

errors = tuple(errors)

__init__

__init__(errors)

Initialize UpstreamError with a list of fetch failures.

Source code in src/ruff_sync/core.py

def __init__(self, errors: Iterable[tuple[URL, BaseException]]) -> None:
    """Initialize UpstreamError with a list of fetch failures."""
    self.errors: Final[tuple[tuple[URL, BaseException], ...]] = tuple(errors)
    error_count = len(self.errors)
    msg = f"❌ {error_count} upstream fetch{'es' if error_count > 1 else ''} failed"
    super().__init__(msg)

DiffContext

Bases: NamedTuple

Context for printing a diff between local and merged configurations.

Source code in src/ruff_sync/core.py

class DiffContext(NamedTuple):
    """Context for printing a diff between local and merged configurations."""

    source_toml_path: pathlib.Path
    source_doc: TOMLDocument
    merged_doc: TOMLDocument
    source_val: object
    merged_val: object

source_toml_path instance-attribute

source_toml_path

source_doc instance-attribute

source_doc

merged_doc instance-attribute

merged_doc

source_val instance-attribute

source_val

merged_val instance-attribute

merged_val

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

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,
    )

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

is_git_url

is_git_url(url)

Return True if the URL should be treated as a git repository.

Source code in src/ruff_sync/core.py

def is_git_url(url: URL) -> bool:
    """Return True if the URL should be treated as a git repository."""
    return str(url).startswith("git@") or url.scheme in ("ssh", "git", "git+ssh")

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

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

download async

download(url, client)

Download a file from a URL and return a StringIO object.

Source code in src/ruff_sync/core.py

async def download(url: URL, client: httpx.AsyncClient) -> StringIO:
    """Download a file from a URL and return a StringIO object."""
    response = await client.get(url)
    response.raise_for_status()
    return StringIO(response.text)

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

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

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

fetch_upstreams_concurrently async

fetch_upstreams_concurrently(
    upstreams, client, branch=DEFAULT_BRANCH, path=None
)

Fetch multiple upstream configurations concurrently.

Uses asyncio.TaskGroup if available (Python 3.11+), otherwise falls back to asyncio.gather.

Parameters:

Name	Type	Description	Default
upstreams	Iterable[URL]	The URLs to fetch.	required
client	AsyncClient	The HTTPX async client to use.	required
branch	str	The default branch for repo-root URLs.	DEFAULT_BRANCH
path	str | None	The directory prefix for pyproject.toml in repo-root URLs.	None

Returns:

Type	Description
list[FetchResult]	A list of FetchResult objects in the same order as the input upstreams.

Raises:

Type	Description
UpstreamError	If one or more upstreams fail to fetch.

Source code in src/ruff_sync/core.py

async def fetch_upstreams_concurrently(
    upstreams: Iterable[URL],
    client: httpx.AsyncClient,
    branch: str = DEFAULT_BRANCH,
    path: str | None = None,
) -> list[FetchResult]:
    """Fetch multiple upstream configurations concurrently.

    Uses asyncio.TaskGroup if available (Python 3.11+), otherwise falls
    back to asyncio.gather.

    Args:
        upstreams: The URLs to fetch.
        client: The HTTPX async client to use.
        branch: The default branch for repo-root URLs.
        path: The directory prefix for pyproject.toml in repo-root URLs.

    Returns:
        A list of FetchResult objects in the same order as the input upstreams.

    Raises:
        UpstreamError: If one or more upstreams fail to fetch.
    """
    upstream_list = list(upstreams)
    if sys.version_info >= (3, 11):
        # Use structured concurrency on Python 3.11+
        tasks: list[asyncio.Task[FetchResult]] = []
        try:
            async with asyncio.TaskGroup() as tg:
                tasks = [
                    tg.create_task(fetch_upstream_config(url, client, branch, path))
                    for url in upstream_list
                ]
            return [t.result() for t in tasks]
        except BaseException as eg:
            if isinstance(eg, (asyncio.CancelledError, KeyboardInterrupt)):
                raise
            # TODO: Use `except*` once Python 3.11+ is the minimum supported version.
            # On Python 3.11+, TaskGroup raises an ExceptionGroup.
            # Catching it as Exception is safe and compatible with Python 3.10 syntax.
            errors = [
                (upstream_list[i], t.exception())
                for i, t in enumerate(tasks)
                if t.done() and t.exception() is not None
            ]
            if errors:
                raise UpstreamError(errors) from eg
            raise
    else:
        # Fallback for Python 3.10
        fetch_tasks = [fetch_upstream_config(url, client, branch, path) for url in upstream_list]
        results = await asyncio.gather(*fetch_tasks, return_exceptions=True)

        errors_list: list[tuple[URL, BaseException]] = []
        fetch_results: list[FetchResult] = []

        for i, res in enumerate(results):
            if isinstance(res, BaseException):
                errors_list.append((upstream_list[i], res))
            elif isinstance(res, FetchResult):
                fetch_results.append(res)
            else:
                msg = f"Unexpected result type from fetch: {type(res)}"
                raise TypeError(msg)

        if errors_list:
            raise UpstreamError(errors_list)

        return fetch_results

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()

serialize_ruff_sync_config

serialize_ruff_sync_config(doc, args)

Serialize the ruff-sync CLI arguments into the TOML document.

Source code in src/ruff_sync/core.py

def serialize_ruff_sync_config(doc: TOMLDocument, args: Arguments) -> None:
    """Serialize the ruff-sync CLI arguments into the TOML document."""
    bad_url = _get_credential_url(args.upstream)
    if bad_url:
        suggested = to_git_url(bad_url)
        suggestion_msg = f" (e.g., {suggested})" if suggested else ""
        LOGGER.warning(
            "⚠️ Upstream URL contains credentials! Refusing to serialize "
            f"[tool.ruff-sync] configuration. Consider using a SSH git URL instead{suggestion_msg}"
            " to avoid leaking credentials."
        )
        return

    ruff_sync_table = _get_or_create_ruff_sync_table(doc)

    # TODO: Consider only saving upstream if it differs from existing config
    if len(args.upstream) == 1:
        ruff_sync_table[ConfKey.UPSTREAM] = str(args.upstream[0])
    else:
        urls_array = tomlkit.array()
        urls_array.multiline(True)
        for url in args.upstream:
            urls_array.append(str(url))
        ruff_sync_table[ConfKey.UPSTREAM] = urls_array

    # Normalize excludes and de-duplicate while preserving order.
    # Only compute and persist excludes when explicitly provided so that
    # DEFAULT_EXCLUDE remains an implicit default and is not serialized.
    if args.exclude != DEFAULT_EXCLUDE:
        normalized_excludes = list(dict.fromkeys(args.exclude))
        exclude_array = tomlkit.array()
        for ex in normalized_excludes:
            exclude_array.append(ex)
        ruff_sync_table[ConfKey.EXCLUDE] = exclude_array

    if args.branch != DEFAULT_BRANCH:
        ruff_sync_table[ConfKey.BRANCH] = args.branch

    if args.path != DEFAULT_PATH and args.path is not None:
        ruff_sync_table[ConfKey.PATH] = args.path

    if args.pre_commit:
        ruff_sync_table[ConfKey.PRE_COMMIT_VERSION_SYNC] = args.pre_commit

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()