Repository: celery/celery
File: celery/utils/nodenames.py (Python)

Code before:
    """
    if isinstance(hostname, Queue):
        return hostname
    return Queue(
        WORKER_DIRECT_QUEUE_FORMAT.format(hostname=hostname),
        WORKER_DIRECT_EXCHANGE,
        hostname,
    )


def nodename(name: str, hostname: str) -> str:
    """Create node name from name/hostname pair."""
    return NODENAME_SEP.join((name, hostname))


def anon_nodename(hostname: str | None = None, prefix: str = 'gen') -> str:
    """Return the nodename for this process (not a worker).

    This is used for e.g. the origin task message field.
    """
    return nodename(''.join([prefix, str(os.getpid())]), hostname or gethostname())


def nodesplit(name: str) -> tuple[None, str] | list[str]:
    """Split node name into tuple of name/hostname."""
    parts: list[str] = name.split(NODENAME_SEP, 1)
    if len(parts) == 1:
        return None, parts[0]
    return parts


def default_nodename(hostname: str) -> str:
    """Return the default nodename for this process."""
    name, host = nodesplit(hostname or '')
    return nodename(name or NODENAME_DEFAULT, host or gethostname())


def node_format(s: str, name: str, **extra: dict) -> str:
    """Format worker node name (name@host.com)."""
    shortname, host = nodesplit(name)
    return host_format(s, host, shortname or NODENAME_DEFAULT, p=name, **extra)


def _fmt_process_index(prefix: str = '', default: str = '0') -> str:
    from .log import current_process_index

    index = current_process_index()
    return f'{prefix}{index}' if index else default


_fmt_process_index_with_prefix = partial(_fmt_process_index, '-', '')

Review comment (suggestion):
```suggestion
    parts = name.split(NODENAME_SEP, 1)
```
This is unnecessary, in general explicitly typing a variable is avoided

Code after:
        WORKER_DIRECT_QUEUE_FORMAT.format(hostname=hostname),
        WORKER_DIRECT_EXCHANGE,
        hostname,
    )


def nodename(name: str, hostname: str) -> str:
    """Create node name from name/hostname pair."""
    return NODENAME_SEP.join((name, hostname))


def anon_nodename(hostname: str | None = None, prefix: str = 'gen') -> str:
    """Return the nodename for this process (not a worker).

    This is used for e.g. the origin task message field.
    """
    return nodename(''.join([prefix, str(os.getpid())]), hostname or gethostname())


def nodesplit(name: str) -> tuple[None, str] | list[str]:
    """Split node name into tuple of name/hostname."""
    parts = name.split(NODENAME_SEP, 1)
    if len(parts) == 1:
        return None, parts[0]
    return parts


def default_nodename(hostname: str) -> str:
    """Return the default nodename for this process."""
    name, host = nodesplit(hostname or '')
    return nodename(name or NODENAME_DEFAULT, host or gethostname())


def node_format(s: str, name: str, **extra: dict) -> str:
    """Format worker node name (name@host.com)."""
    shortname, host = nodesplit(name)
    return host_format(s, host, shortname or NODENAME_DEFAULT, p=name, **extra)


def _fmt_process_index(prefix: str = '', default: str = '0') -> str:
    from .log import current_process_index

    index = current_process_index()
    return f'{prefix}{index}' if index else default


_fmt_process_index_with_prefix = partial(_fmt_process_index, '-', '')


def host_format(s: str, host: str | None = None, name: str | None = None, **extra: dict) -> str:
    """Format host %x abbreviations."""