DM-44230: Add Slack error reporting

CHANGELOG.md

@@ -7,6 +7,22 @@ Find changes for the upcoming release in the project's [changelog.d directory](h
 
 <!-- scriv-insert-here -->
 
+<a id='changelog-2.0.0'></a>
+## 2.0.0 (2024-05-07)
+
+### Backwards-incompatible changes
+
+- The `GET /fastapi-bootcamp/astroplan/observers` endpoint now uses pagination.
+
+### New features
+
+- Demonstrate `SlackException` in the `POST /fastapi-bootcamp/error-demo` endpoint.
+- Demonstrate custom FastAPI dependencies in the `GET /fastapi-bootcamp/dependency-demo` endpoint.
+
+### Other changes
+
+Minor improvements to handler docs.
+
 <a id='changelog-1.0.0'></a>
 ## 1.0.0 (2024-04-30)
 

pyproject.toml

@@ -168,6 +168,9 @@ select = ["ALL"]
 "src/fastapibootcamp/handlers/external.py" = [
     "ERA001",  # Allow some commented code for documentation
 ]
+"src/fastapibootcamp/dependencies/singletondependency.py" = [
+    "S311",    # Allow use of random in this module
+]
 "tests/**" = [
     "C901",    # tests are allowed to be complex, sometimes that's convenient
     "D101",    # tests don't need docstrings

src/fastapibootcamp/config.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from pydantic import Field
+from pydantic import Field, HttpUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from safir.logging import LogLevel, Profile
 
@@ -26,6 +26,13 @@ class Config(BaseSettings):
         LogLevel.INFO, title="Log level of the application's logger"
     )
 
+    slack_webhook_url: HttpUrl | None = Field(
+        None,
+        description=(
+            "Webhook URL for sending error messages to a Slack channel."
+        ),
+    )
+
     clear_iers_on_startup: bool = Field(
         False,
         title="Clear IERS cache on application startup",

src/fastapibootcamp/dependencies/pagination.py

@@ -0,0 +1,86 @@
+"""A functional dependency that adds a pagination query string parameter to a
+FastAPI path operation.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Annotated
+
+from fastapi import Query
+
+__all__ = ["Pagination", "SortOrder", "pagination_dependency"]
+
+
+class SortOrder(str, Enum):
+    """Sort order."""
+
+    asc = "asc"
+    desc = "desc"
+
+
+@dataclass
+class Pagination:
+    """Pagination parameters."""
+
+    page: int
+    """The requested page number."""
+
+    limit: int
+    """The requested number of items per page."""
+
+    order: SortOrder
+    """The requested sort order."""
+
+    @property
+    def query_params(self) -> dict[str, str]:
+        """Get the URL query string parameters for this page.
+
+        This can be used to build a URL with a query string for the current
+        page.
+        """
+        return {
+            "page": str(self.page),
+            "limit": str(self.limit),
+            "order": self.order.value,
+        }
+
+
+async def pagination_dependency(
+    page: Annotated[
+        int, Query(ge=1, title="Pagination page.", examples=[1, 2, 3])
+    ] = 1,
+    limit: Annotated[
+        int, Query(title="Max number of items in page.", examples=[10, 20, 30])
+    ] = 10,
+    order: Annotated[
+        SortOrder,
+        Query(title="Sort order.", examples=[SortOrder.asc, SortOrder.desc]),
+    ] = SortOrder.asc,
+) -> Pagination:
+    """Add pagination query string parameters to a FastAPI path operation.
+
+    This dependency adds three query string parameters to a FastAPI path
+    operation: `page`, `limit`, and `order`.
+
+    Note that this sets up "offset" pagination, which is simple to implement
+    for this demo. With a real database, you may want to look into "cursor"
+    based pagination for better performance and reliability with dynamic data.
+
+    Parameters
+    ----------
+    page
+        The page number.
+    limit
+        The number of items to return per page.
+    order
+        The sort order.
+
+    Returns
+    -------
+    Pagination
+        A container with the `page`, `limit`, and `order` query string
+        parameters.
+    """
+    return Pagination(page=page, limit=limit, order=order)

src/fastapibootcamp/dependencies/singletondependency.py

@@ -0,0 +1,80 @@
+"""A class-based FastAPI dependency for demonstration purposes."""
+
+from __future__ import annotations
+
+import random
+
+from ..exceptions import DemoInternalError
+
+__all__ = ["example_singleton_dependency", "ExampleSingletonDependency"]
+
+
+class ExampleSingletonDependency:
+    """A stateful FastAPI dependency for demonstration purposes.
+
+    See lesson 6 in src/fastapibootcamp/handlers/example.py for usage.
+    """
+
+    def __init__(self) -> None:
+        # For this demo we're just using a semi-random string as the state. In
+        # real applications, this could be a database connection, a client
+        # to a remote service, etc. This "state" is reused over the life of
+        # this application instance (i.e. a Kubernetes pod). It's not shared
+        # between instances, though.
+        self._state: str | None = None
+
+    async def init(self) -> None:
+        """Initialize the dependency.
+
+        This initialization is called in main.py in the lifespan context
+        manager.
+        """
+        self._state = f"{random.choice(ADJECTIVES)} {random.choice(ANIMALS)}"
+
+    async def __call__(self) -> str:
+        """Provide the dependency.
+
+        This gets called by the fastapi Depends() function when your
+        path operation function is called.
+        """
+        if self._state is None:
+            raise DemoInternalError(
+                "ExampleSingletonDependency not initialized."
+            )
+
+        return self._state
+
+    async def aclose(self) -> None:
+        """Clean up the dependency.
+
+        If needed, this method is called when the application is shutting down
+        to close connections, etc.. This is called in the lifespan context
+        manager in main.py
+        """
+        self._state = None
+
+
+# This is the singleton instance of the dependency that's referenced in path
+# operation functions with the fastapi.Depends() function. Note that it needs
+# to be initialized before it can be used. This is done in the lifespan context
+# manager in main.py. Another option is to initialize it on the first use.
+example_singleton_dependency = ExampleSingletonDependency()
+
+
+ADJECTIVES = [
+    "speedy",
+    "ponderous",
+    "furious",
+    "careful",
+    "mammoth",
+    "crafty",
+]
+
+ANIMALS = [
+    "cat",
+    "dog",
+    "sloth",
+    "snail",
+    "rabbit",
+    "turtle",
+]

src/fastapibootcamp/domain/models.py

@@ -4,13 +4,15 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any
+from typing import Any, Self
 
 from astroplan import Observer as AstroplanObserver
 from astropy.coordinates import AltAz, Angle, SkyCoord
 from astropy.time import Time
 
-__all__ = ["Observer", "TargetObservability"]
+from ..dependencies.pagination import Pagination
+
+__all__ = ["Observer", "ObserversPage", "TargetObservability"]
 
 # The domain layer is where your application's core business logic resides.
 # In this demo, the domain is built around the Astroplan library and its
@@ -42,6 +44,30 @@ def __init__(
         self.local_timezone = local_timezone
 
 
+@dataclass(kw_only=True)
+class ObserversPage:
+    """A paged collection of observer items.
+
+    Parameters
+    ----------
+    observers
+        The observers on this page.
+    total
+        The total number of observers across all pages.
+    pagination
+        The current page.
+    """
+
+    observers: list[Observer]
+    """The observers in this page."""
+
+    total: int
+    """The total number of observers across all pages."""
+
+    pagination: Pagination
+    """The current page."""
+
+
 @dataclass(kw_only=True)
 class TargetObservability:
     """The observability of a target for an observer."""
@@ -64,7 +90,7 @@ def compute(
         observer: Observer,
         target: SkyCoord,
         time: datetime,
-    ) -> TargetObservability:
+    ) -> Self:
         """Compute the observability of a target for an observer."""
         astropy_time = Time(time)
         is_up = observer.target_is_up(astropy_time, target)

src/fastapibootcamp/exceptions.py

@@ -1,12 +1,47 @@
 """Exceptions for the FastAPI Bootcamp app."""
 
 from safir.fastapi import ClientRequestError
+from safir.slack.blockkit import SlackException, SlackMessage, SlackTextField
 
 __all__ = [
     "ObserverNotFoundError",
 ]
 
 
+class DemoInternalError(SlackException):
+    """Raised when a demo internal error occurs.
+
+    A `SlackException` subclass is a custom exception provided by Safir that
+    sends a message to a Slack channel, though the webhook URL, when it's
+    raised. You can use the `to_slack` method to format that Slack message.
+
+    SlackException subclasses are used by internal application errors
+    (i.e., a request *should* have worked given the validated inputs, but
+    something unexpected happend). On the other hand, if the request can't
+    be completed because the user inputs are invalid, use the
+    `ClientRequestError` subclass instead (see below).
+
+    In other words, use `SlackException` for 500-type erros and
+    `ClientRequestError` for 400-type errors.
+
+    Note: to use SlackException, you need to set up the SlackRouteErrorHandler
+    middleware in the FastAPI application. See `src/fastapibootcamp/main.py`.
+    """
+
+    def __init__(self, msg: str, custom_data: str | None = None) -> None:
+        """Initialize the exception."""
+        super().__init__(msg)
+        self.custom_data = custom_data
+
+    def to_slack(self) -> SlackMessage:
+        message = super().to_slack()
+        if self.custom_data:
+            message.fields.append(
+                SlackTextField(heading="Data", text=self.custom_data)
+            )
+        return message
+
+
 class ObserverNotFoundError(ClientRequestError):
     """Raised when an observing site is not found."""