Repository: home-assistant/core
File: homeassistant/components/ntfy/notify.py (Python)

Code before:
            params["delay"] = f"{delay.total_seconds()}s"
            if params.get("email"):
                raise ServiceValidationError(
                    translation_domain=DOMAIN,
                    translation_key="delay_no_email",
                )
            if params.get("call"):
                raise ServiceValidationError(
                    translation_domain=DOMAIN,
                    translation_key="delay_no_call",
                )
        if file := params.pop(ATTR_ATTACH_FILE, None):
            if params.get(ATTR_ATTACH) is not None:
                raise ServiceValidationError(
                    translation_domain=DOMAIN,
                    translation_key="attach_url_xor_local",
                )
            media_content_id: str = file["media_content_id"]
            if media_content_id.startswith("media-source://camera/"):
                entity_id = media_content_id.removeprefix("media-source://camera/")
                img = await camera.async_get_image(self.hass, entity_id)
                attachment = img.content
            elif media_content_id.startswith("media-source://image/"):
                entity_id = media_content_id.removeprefix("media-source://image/")
                if (
                    entity := self.hass.data[IMAGE_DATA_COMPONENT].get_entity(entity_id)
                ) is None:
                    raise HomeAssistantError(
                        translation_domain=DOMAIN,
                        translation_key="image_source_not_found",
                    )
                attachment = await entity.async_image()
            else:
                media = await async_resolve_media(
                    self.hass, file["media_content_id"], None
                )

                if media.path is None:
                    raise ServiceValidationError(
                        translation_domain=DOMAIN,
                        translation_key="media_source_not_supported",
                    )
                async with aiofiles.open(media.path, mode="rb") as f:
                    attachment = await f.read()
                params.setdefault(ATTR_FILENAME, media.path.name)

        msg = Message(topic=self.topic, **params)
        try:
            await self.ntfy.publish(msg, attachment)
        except NtfyUnauthorizedAuthenticationError as e:
            self.config_entry.async_start_reauth(self.hass)

Review comment (suggestion):
We're not allowed to access `hass.data` of another integration directly. If an integration wants to expose a feature to other integrations it must do so via a top level function that can be imported and used by other integrations.

Why don't we use `image.async_get_image`?

https://github.com/home-assistant/core/blob/7a696935eded247fd36cc2458c59d8379357ff32/homeassistant/components/image/__init__.py#L127-L138

Code after:
    platform.async_register_entity_service(
        SERVICE_PUBLISH,
        SERVICE_PUBLISH_SCHEMA,
        "publish",
        description_placeholders={
            "markdown_guide_url": "https://www.markdownguide.org/basic-syntax/",
            "emoji_reference_url": "https://docs.ntfy.sh/emojis/",
        },
    )

    platform.async_register_entity_service(
        SERVICE_CLEAR,
        SERVICE_CLEAR_DELETE_SCHEMA,
        "clear",
    )

    platform.async_register_entity_service(
        SERVICE_DELETE,
        SERVICE_CLEAR_DELETE_SCHEMA,
        "delete",
    )


class NtfyNotifyEntity(NtfyBaseEntity, NotifyEntity):
    """Representation of a ntfy notification entity."""

    entity_description = NotifyEntityDescription(
        key="publish",
        translation_key="publish",
        name=None,
    )
    _attr_supported_features = NotifyEntityFeature.TITLE

    async def async_send_message(self, message: str, title: str | None = None) -> None:
        """Publish a message to a topic."""
        await self.publish(message=message, title=title)

    async def publish(self, **kwargs: Any) -> None:
        """Publish a message to a topic."""
        attachment = None
        params: dict[str, Any] = kwargs
        delay: timedelta | None = params.get("delay")
        if delay:
            params["delay"] = f"{delay.total_seconds()}s"
            if params.get("email"):
                raise ServiceValidationError(
                    translation_domain=DOMAIN,
                    translation_key="delay_no_email",
                )
            if params.get("call"):
                raise ServiceValidationError(