ical.calendar_stream

  1"""The core, a collection of Calendar and Scheduling objects.
  2
  3This is an example of parsing an ics file as a stream of calendar objects:
  4```python
  5from pathlib import Path
  6from ical.calendar_stream import IcsCalendarStream
  7
  8filename = Path("example/calendar.ics")
  9with filename.open() as ics_file:
 10    stream = IcsCalendarStream.from_ics(ics_file.read())
 11    print("File contains %s calendar(s)", len(stream.calendars))
 12```
 13
 14You can encode a calendar stream as ics content calling the `ics()` method on
 15the `IcsCalendarStream`:
 16
 17```python
 18from pathlib import Path
 19
 20filename = Path("/tmp/output.ics")
 21with filename.open(mode="w") as ics_file:
 22    ics_file.write(stream.ics())
 23```
 24
 25Fetching a remote ics file without blocking the event loop requires the
 26optional `ical[async]` extra (which installs `aiohttp`):
 27
 28```python
 29from ical.calendar_stream import IcsCalendarStream
 30
 31stream = await IcsCalendarStream.from_url("https://example.com/calendar.ics")
 32```
 33
 34"""
 35
 36# ty: allow-any-generics
 37
 38from __future__ import annotations
 39
 40import logging
 41from typing import TYPE_CHECKING
 42
 43from pydantic import Field, field_serializer
 44
 45from .calendar import Calendar
 46from .component import ComponentModel
 47from .parsing.component import encode_content, parse_content
 48from .types.data_types import serialize_field
 49from .exceptions import CalendarFetchError, CalendarParseError
 50from pydantic import ConfigDict
 51
 52if TYPE_CHECKING:
 53    import aiohttp
 54
 55_LOGGER = logging.getLogger(__name__)
 56
 57__all__ = ["CalendarStream", "IcsCalendarStream"]
 58
 59_ASYNC_EXTRA_ERROR = (
 60    "The aiohttp library is required for async URL fetching. Install it "
 61    "with the optional extra: `pip install ical[async]`"
 62)
 63
 64
 65class CalendarStream(ComponentModel):
 66    """A container that is a collection of calendaring information.
 67
 68    This object supports parsing an rfc5545 calendar file, but does not
 69    support encoding. See `IcsCalendarStream` instead for encoding ics files.
 70    """
 71
 72    calendars: list[Calendar] = Field(alias="vcalendar", default_factory=list)
 73
 74    @classmethod
 75    def from_ics(cls, content: str) -> "CalendarStream":
 76        """Factory method to create a new instance from an rfc5545 iCalendar content."""
 77        components = parse_content(content)
 78        result: dict[str, list] = {"vcalendar": []}
 79        for component in components:
 80            result.setdefault(component.name, [])
 81            result[component.name].append(component.as_dict())
 82        _LOGGER.debug("Parsing object %s", result)
 83        return cls(**result)
 84
 85    @classmethod
 86    async def from_url(
 87        cls, url: str, session: "aiohttp.ClientSession | None" = None
 88    ) -> "CalendarStream":
 89        """Async factory method to fetch and parse an rfc5545 iCalendar url.
 90
 91        This avoids blocking the event loop while fetching a remote ics file
 92        (e.g. a calendar subscription url), which is useful for applications
 93        such as Home Assistant that are built on asyncio. Parsing the fetched
 94        content happens synchronously since it is fast and CPU-bound.
 95
 96        This requires the optional `aiohttp` dependency, installable with the
 97        `ical[async]` extra. An existing `aiohttp.ClientSession` may be
 98        passed in to reuse connection pooling; otherwise a session is created
 99        and closed automatically for this single request.
100        """
101        try:
102            import aiohttp  # noqa: PLC0415
103        except ImportError as err:
104            raise ImportError(_ASYNC_EXTRA_ERROR) from err
105
106        async def _fetch(active_session: aiohttp.ClientSession) -> str:
107            try:
108                async with active_session.get(url) as response:
109                    response.raise_for_status()
110                    return await response.text()
111            except aiohttp.ClientError as err:
112                raise CalendarFetchError(
113                    f"Failed to fetch calendar from url {url!r}: {err}"
114                ) from err
115
116        if session is not None:
117            content = await _fetch(session)
118        else:
119            async with aiohttp.ClientSession() as owned_session:
120                content = await _fetch(owned_session)
121        return cls.from_ics(content)
122
123    def ics(self) -> str:
124        """Encode the calendar stream as an rfc5545 iCalendar Stream content."""
125        return encode_content(self.__encode_component_root__().components)
126
127
128class IcsCalendarStream(CalendarStream):
129    """A calendar stream that supports parsing and encoding ICS."""
130
131    @classmethod
132    def calendar_from_ics(cls, content: str) -> Calendar:
133        """Load a single calendar from an ics string."""
134        stream = cls.from_ics(content)
135        return cls._single_calendar(stream)
136
137    @classmethod
138    async def calendar_from_url(
139        cls, url: str, session: "aiohttp.ClientSession | None" = None
140    ) -> Calendar:
141        """Async convenience method to fetch and load a single calendar from a url.
142
143        See `from_url` for details on the optional `ical[async]` dependency
144        and the `session` argument.
145        """
146        stream = await cls.from_url(url, session=session)
147        return cls._single_calendar(stream)
148
149    @staticmethod
150    def _single_calendar(stream: "CalendarStream") -> Calendar:
151        """Return the single calendar in the stream, or raise an error."""
152        if len(stream.calendars) == 1:
153            return stream.calendars[0]
154        if len(stream.calendars) == 0:
155            return Calendar()
156        raise CalendarParseError("Calendar Stream had more than one calendar")
157
158    @classmethod
159    def calendar_to_ics(cls, calendar: Calendar) -> str:
160        """Serialize a calendar as an ICS stream."""
161        stream = cls(vcalendar=[calendar])
162        return stream.ics()
163
164    model_config = ConfigDict(
165        validate_assignment=True,
166        populate_by_name=True,
167    )
168    serialize_fields = field_serializer("*")(serialize_field)  # type: ignore[pydantic-field]