Function hacs_repositories_add

custom_components/hacs/websocket/repositories.py:143–193

Add custom repositoriy.

(
    hass: HomeAssistant,
    connection: websocket_api.ActiveConnection,
    msg: dict[str, Any],
)

@websocket_api.require_admin
@websocket_api.async_response
async def hacs_repositories_add(
    hass: HomeAssistant,
    connection: websocket_api.ActiveConnection,
    msg: dict[str, Any],
) -> None:
    """Add custom repositoriy."""
    hacs: HacsBase = hass.data.get(DOMAIN)
    repository = regex.extract_repository_from_url(msg["repository"])
    category = msg["category"]

    if repository is None:
        return

    if repository in hacs.common.skip:
        hacs.common.skip.remove(repository)

    if renamed := hacs.common.renamed_repositories.get(repository):
        repository = renamed

    if category not in hacs.common.categories:
        hacs.log.error("%s is not a valid category for %s", category, repository)

    elif not hacs.repositories.get_by_full_name(repository):
        try:
            await hacs.async_register_repository(
                repository_full_name=repository,
                category=category,
            )

        except (
            BaseException  # lgtm [py/catch-base-exception] pylint: disable=broad-except
        ) as exception:
            hacs.async_dispatch(
                HacsDispatchEvent.ERROR,
                {
                    "action": "add_repository",
                    "exception": str(sys.exc_info()[0].__name__),
                    "message": str(exception),
                },
            )

    else:
        hacs.async_dispatch(
            HacsDispatchEvent.ERROR,
            {
                "action": "add_repository",
                "message": f"Repository '{repository}' exists in the store.",
            },
        )

    connection.send_message(websocket_api.result_message(msg["id"], {}))


@websocket_api.websocket_command(