Merge pull request #101 from lsst-sqre/tickets/DM-37516

DM-37516: Update typing to avoid deprecations

src/vocutouts/actors.py

@@ -34,7 +34,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, List
+from typing import Any
 
 import dramatiq
 import structlog
@@ -55,9 +55,9 @@
 @dramatiq.actor(queue_name="cutout", max_retries=1, store_results=True)
 def cutout(
     job_id: str,
-    dataset_ids: List[str],
-    stencils: List[Dict[str, Any]],
-) -> List[Dict[str, str]]:
+    dataset_ids: list[str],
+    stencils: list[dict[str, Any]],
+) -> list[dict[str, str]]:
     """Stub for a circle cutout.
 
     This is only a stub, existing to define the actor in the Dramatiq broker
@@ -68,19 +68,19 @@ def cutout(
     ----------
     job_id : `str`
         The UWS job ID, used as the key for storing results.
-    dataset_ids : List[`str`]
+    dataset_ids : list[`str`]
         The data objects on which to perform cutouts.  These are opaque
         identifiers passed as-is to the backend.  The user will normally
         discover them via some service such as ObsTAP.
-    stencils : List[Dict[`str`, Any]]
+    stencils : list[dict[`str`, Any]]
         Serialized stencils for the cutouts to perform.  These are
         JSON-serializable (a requirement for Dramatiq) representations of the
         `~vocutouts.models.stencils.Stencil` objects corresponding to the
         user's request.
 
     Returns
     -------
-    result : List[Dict[`str`, `str`]]
+    result : list[dict[`str`, `str`]]
         The results of the job.  This must be a list of dict representations
         of `~vocutouts.uws.models.JobResult` objects.
 
@@ -117,7 +117,7 @@ def job_started(job_id: str, message_id: str, start_time: str) -> None:
 
 @dramatiq.actor(queue_name="uws", priority=10)
 def job_completed(
-    message: Dict[str, Any], result: List[Dict[str, str]]
+    message: dict[str, Any], result: list[dict[str, str]]
 ) -> None:
     """Wrapper around the UWS function to mark a job as completed."""
     logger = structlog.get_logger(config.logger_name)
@@ -127,7 +127,7 @@ def job_completed(
 
 
 @dramatiq.actor(queue_name="uws", priority=20)
-def job_failed(message: Dict[str, Any], exception: Dict[str, str]) -> None:
+def job_failed(message: dict[str, Any], exception: dict[str, str]) -> None:
     """Wrapper around the UWS function to mark a job as errored."""
     logger = structlog.get_logger(config.logger_name)
     job_id = message["args"][0]

src/vocutouts/cli.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-from typing import Optional
-
 import click
 import structlog
 import uvicorn
@@ -27,7 +25,7 @@ def main() -> None:
 @main.command()
 @click.argument("topic", default=None, required=False, nargs=1)
 @click.pass_context
-def help(ctx: click.Context, topic: Optional[str]) -> None:
+def help(ctx: click.Context, topic: str | None) -> None:
     """Show help for any command."""
     # The help command implementation is taken from
     # https://www.burgundywall.com/post/having-click-help-subcommand

src/vocutouts/config.py

@@ -4,7 +4,6 @@
 
 import os
 from dataclasses import dataclass
-from typing import Optional
 
 from .uws.config import UWSConfig
 
@@ -29,7 +28,7 @@ class Configuration:
     is mandatory.
     """
 
-    database_password: Optional[str] = os.getenv("CUTOUT_DATABASE_PASSWORD")
+    database_password: str | None = os.getenv("CUTOUT_DATABASE_PASSWORD")
     """The password for the UWS job database.
 
     Set with the ``CUTOUT_DATABASE_PASSWORD`` environment variable.
@@ -64,7 +63,7 @@ class Configuration:
     mandatory.
     """
 
-    redis_password: Optional[str] = os.getenv("CUTOUT_REDIS_PASSWORD")
+    redis_password: str | None = os.getenv("CUTOUT_REDIS_PASSWORD")
     """Password for the Redis server used by Dramatiq.
 
     Set with the ``CUTOUT_REDIS_PASSWORD`` environment variable.

src/vocutouts/exceptions.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-from typing import List
-
 from .uws.exceptions import ParameterError
 from .uws.models import JobParameter
 
@@ -13,7 +11,7 @@
 class InvalidCutoutParameterError(ParameterError):
     """The parameters for the cutout were invalid."""
 
-    def __init__(self, message: str, params: List[JobParameter]) -> None:
+    def __init__(self, message: str, params: list[JobParameter]) -> None:
         detail = "\n".join(f"{p.parameter_id}={p.value}" for p in params)
         super().__init__(message, detail)
         self.params = params

src/vocutouts/handlers/external.py

@@ -5,7 +5,7 @@
 application knows the job parameters.
 """
 
-from typing import List, Literal, Optional
+from typing import Literal, Optional
 
 from fastapi import APIRouter, Depends, Form, Query, Request, Response
 from fastapi.responses import PlainTextResponse, RedirectResponse
@@ -118,9 +118,9 @@ async def get_capabilities(request: Request) -> Response:
 
 
 async def _sync_request(
-    params: List[JobParameter],
+    params: list[JobParameter],
     user: str,
-    runid: Optional[str],
+    runid: str | None,
     uws_factory: UWSFactory,
     logger: BoundLogger,
 ) -> Response:
@@ -193,15 +193,15 @@ async def _sync_request(
 )
 async def get_sync(
     request: Request,
-    id: List[str] = Query(
+    id: list[str] = Query(
         ...,
         title="Source ID",
         description=(
             "Identifiers of images from which to make a cutout. This"
             " parameter is mandatory."
         ),
     ),
-    pos: Optional[List[str]] = Query(
+    pos: Optional[list[str]] = Query(
         None,
         title="Cutout positions",
         description=(
@@ -213,7 +213,7 @@ async def get_sync(
             " numbers expressed as strings."
         ),
     ),
-    circle: Optional[List[str]] = Query(
+    circle: Optional[list[str]] = Query(
         None,
         title="Cutout circle positions",
         description=(
@@ -223,7 +223,7 @@ async def get_sync(
             " strings and separated by spaces."
         ),
     ),
-    polygon: Optional[List[str]] = Query(
+    polygon: Optional[list[str]] = Query(
         None,
         title="Cutout polygon positions",
         description=(
@@ -274,15 +274,15 @@ async def get_sync(
 )
 async def post_sync(
     request: Request,
-    id: Optional[List[str]] = Form(
+    id: Optional[list[str]] = Form(
         None,
         title="Source ID",
         description=(
             "Identifiers of images from which to make a cutout. This"
             " parameter is mandatory."
         ),
     ),
-    pos: Optional[List[str]] = Form(
+    pos: Optional[list[str]] = Form(
         None,
         title="Cutout positions",
         description=(
@@ -294,7 +294,7 @@ async def post_sync(
             " numbers expressed as strings."
         ),
     ),
-    circle: Optional[List[str]] = Form(
+    circle: Optional[list[str]] = Form(
         None,
         title="Cutout circle positions",
         description=(
@@ -304,7 +304,7 @@ async def post_sync(
             " strings and separated by spaces."
         ),
     ),
-    polygon: Optional[List[str]] = Form(
+    polygon: Optional[list[str]] = Form(
         None,
         title="Cutout polygon positions",
         description=(
@@ -324,7 +324,7 @@ async def post_sync(
             " with specific larger operations."
         ),
     ),
-    params: List[JobParameter] = Depends(uws_post_params_dependency),
+    params: list[JobParameter] = Depends(uws_post_params_dependency),
     user: str = Depends(auth_dependency),
     uws_factory: UWSFactory = Depends(uws_dependency),
     logger: BoundLogger = Depends(auth_logger_dependency),
@@ -346,15 +346,15 @@ async def post_sync(
 )
 async def create_job(
     request: Request,
-    id: Optional[List[str]] = Form(
+    id: Optional[list[str]] = Form(
         None,
         title="Source ID",
         description=(
             "Identifiers of images from which to make a cutout. This"
             " parameter is mandatory."
         ),
     ),
-    pos: Optional[List[str]] = Form(
+    pos: Optional[list[str]] = Form(
         None,
         title="Cutout positions",
         description=(
@@ -366,7 +366,7 @@ async def create_job(
             " numbers expressed as strings."
         ),
     ),
-    circle: Optional[List[str]] = Form(
+    circle: Optional[list[str]] = Form(
         None,
         title="Cutout circle positions",
         description=(
@@ -376,7 +376,7 @@ async def create_job(
             " strings and separated by spaces."
         ),
     ),
-    polygon: Optional[List[str]] = Form(
+    polygon: Optional[list[str]] = Form(
         None,
         title="Cutout polygon positions",
         description=(
@@ -399,7 +399,7 @@ async def create_job(
             " with specific larger operations."
         ),
     ),
-    params: List[JobParameter] = Depends(uws_post_params_dependency),
+    params: list[JobParameter] = Depends(uws_post_params_dependency),
     user: str = Depends(auth_dependency),
     uws_factory: UWSFactory = Depends(uws_dependency),
     logger: BoundLogger = Depends(auth_logger_dependency),

src/vocutouts/models/parameters.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import List
 
 from ..exceptions import InvalidCutoutParameterError
 from ..uws.models import JobParameter
@@ -14,21 +13,21 @@
 class CutoutParameters:
     """The parameters to a cutout request."""
 
-    ids: List[str]
+    ids: list[str]
     """The dataset IDs on which to operate."""
 
-    stencils: List[Stencil]
+    stencils: list[Stencil]
     """The cutout stencils to apply."""
 
     @classmethod
     def from_job_parameters(
-        cls, params: List[JobParameter]
+        cls, params: list[JobParameter]
     ) -> CutoutParameters:
         """Convert generic UWS parameters to the iamge cutout parameters.
 
         Parameters
         ----------
-        params : List[`vocutouts.uws.models.JobParameter`]
+        params : list[`vocutouts.uws.models.JobParameter`]
             Generic input job parameters.
 
         Returns

src/vocutouts/models/stencils.py

@@ -4,12 +4,12 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Dict, Tuple
+from typing import Any
 
 from astropy import units as u
 from astropy.coordinates import Angle, SkyCoord
 
-Range = Tuple[float, float]
+Range = tuple[float, float]
 
 
 class Stencil(ABC):
@@ -21,7 +21,7 @@ def from_string(cls, params: str) -> Stencil:
         """Parse a string representation of stencil parameters to an object."""
 
     @abstractmethod
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         """Convert the stencil to a JSON-serializable form for queuing."""
 
 
@@ -40,7 +40,7 @@ def from_string(cls, params: str) -> CircleStencil:
             radius=Angle(radius * u.degree),
         )
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         return {
             "type": "circle",
             "center": {
@@ -78,7 +78,7 @@ def from_string(cls, params: str) -> PolygonStencil:
         vertices = SkyCoord(ras * u.degree, decs * u.degree, frame="icrs")
         return cls(vertices=vertices)
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         return {
             "type": "polygon",
             "vertices": [(v.ra.degree, v.dec.degree) for v in self.vertices],
@@ -100,7 +100,7 @@ def from_string(cls, params: str) -> RangeStencil:
             dec=(dec_min, dec_max),
         )
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         return {
             "type": "range",
             "ra": self.ra,

src/vocutouts/policy.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import List
 
 from dramatiq import Actor, Message
 from structlog.stdlib import BoundLogger
@@ -80,7 +79,7 @@ def validate_execution_duration(
     ) -> int:
         return job.execution_duration
 
-    def validate_params(self, params: List[JobParameter]) -> None:
+    def validate_params(self, params: list[JobParameter]) -> None:
         try:
             cutout_params = CutoutParameters.from_job_parameters(params)
         except InvalidCutoutParameterError as e:

src/vocutouts/uws/config.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Optional
 
 __all__ = ["UWSConfig"]
 
@@ -49,10 +48,10 @@ class encapsulates the configuration of the UWS component that may vary by
     with this email.
     """
 
-    database_password: Optional[str] = None
+    database_password: str | None = None
     """Password for the database."""
 
-    redis_password: Optional[str] = None
+    redis_password: str | None = None
     """Password for the Redis server used by Dramatiq."""
 
     url_lifetime: int = 15 * 60