refactor: use new Python syntax for Optional typing (#1624)

Author: raphael0202

robotoff/app/api.py

@@ -10,7 +10,7 @@
 import uuid
 from collections import defaultdict
 from pathlib import Path
-from typing import Literal, Optional, cast
+from typing import Literal, cast
 
 import falcon
 import orjson
@@ -119,9 +119,9 @@ def get_server_type_from_req(
 COUNTRY_NAME_TO_ENUM = {COUNTRY_CODE_TO_NAME[item]: item for item in Country}
 
 
-def get_countries_from_req(req: falcon.Request) -> Optional[list[Country]]:
+def get_countries_from_req(req: falcon.Request) -> list[Country] | None:
     """Parse `countries` query string from request."""
-    countries_str: Optional[list[str]] = req.get_param_as_list("countries")
+    countries_str: list[str] | None = req.get_param_as_list("countries")
     if countries_str is None:
         return None
 
@@ -202,12 +202,12 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
         response: JSONType = {}
         count: int = req.get_param_as_int("count", min_value=1, default=25)
         page: int = req.get_param_as_int("page", min_value=1, default=1)
-        keep_types: Optional[list[str]] = req.get_param_as_list(
+        keep_types: list[str] | None = req.get_param_as_list(
             "insight_types", required=False
         )
-        barcode: Optional[str] = normalize_req_barcode(req.get_param("barcode"))
-        annotated: Optional[bool] = req.get_param_as_bool("annotated")
-        annotation: Optional[int] = req.get_param_as_int("annotation")
+        barcode: str | None = normalize_req_barcode(req.get_param("barcode"))
+        annotated: bool | None = req.get_param_as_bool("annotated")
+        annotation: int | None = req.get_param_as_int("annotation")
         value_tag: str = req.get_param("value_tag")
         brands = req.get_param_as_list("brands") or None
         predictor = req.get_param("predictor")
@@ -272,8 +272,8 @@ class RandomInsightResource:
     def on_get(self, req: falcon.Request, resp: falcon.Response):
         response: JSONType = {}
 
-        insight_type: Optional[str] = req.get_param("type")
-        value_tag: Optional[str] = req.get_param("value_tag")
+        insight_type: str | None = req.get_param("type")
+        value_tag: str | None = req.get_param("value_tag")
         count: int = req.get_param_as_int("count", min_value=1, default=25)
         predictor = req.get_param("predictor")
         server_type = get_server_type_from_req(req)
@@ -303,7 +303,7 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
         resp.media = response
 
 
-def parse_auth(req: falcon.Request) -> Optional[OFFAuthentication]:
+def parse_auth(req: falcon.Request) -> OFFAuthentication | None:
     session_cookie = req.get_cookie_values("session")
 
     if session_cookie:
@@ -917,13 +917,13 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
         count: int = req.get_param_as_int(
             "count", min_value=1, max_value=2000, default=25
         )
-        type_: Optional[str] = req.get_param("type")
-        barcode: Optional[str] = normalize_req_barcode(req.get_param("barcode"))
-        value: Optional[str] = req.get_param("value")
-        taxonomy_value: Optional[str] = req.get_param("taxonomy_value")
-        min_confidence: Optional[float] = req.get_param_as_float("min_confidence")
+        type_: str | None = req.get_param("type")
+        barcode: str | None = normalize_req_barcode(req.get_param("barcode"))
+        value: str | None = req.get_param("value")
+        taxonomy_value: str | None = req.get_param("taxonomy_value")
+        min_confidence: float | None = req.get_param_as_float("min_confidence")
         random: bool = req.get_param_as_bool("random", default=False)
-        annotated: Optional[bool] = req.get_param_as_bool("annotated")
+        annotated: bool | None = req.get_param_as_bool("annotated")
 
         if type_ is None and (value is not None or taxonomy_value is not None):
             raise falcon.HTTPBadRequest(
@@ -1031,7 +1031,7 @@ def on_put(self, req: falcon.Request, resp: falcon.Response, logo_id: int):
             )
 
         with db.atomic():
-            logo: Optional[LogoAnnotation] = LogoAnnotation.get_or_none(id=logo_id)
+            logo: LogoAnnotation | None = LogoAnnotation.get_or_none(id=logo_id)
             if logo is None:
                 resp.status = falcon.HTTP_404
                 return
@@ -1201,7 +1201,7 @@ def on_post(self, req: falcon.Request, resp: falcon.Response):
 
 class ANNResource:
     def on_get(
-        self, req: falcon.Request, resp: falcon.Response, logo_id: Optional[int] = None
+        self, req: falcon.Request, resp: falcon.Response, logo_id: int | None = None
     ):
         """Search for nearest neighbors of:
         - a random logo (if logo_id not provided)
@@ -1311,9 +1311,9 @@ def on_get(self, req: falcon.Request, resp: falcon.Response, barcode: str):
         # hash of the IPs as a backup.
         device_id = device_id_from_request(req)
 
-        auth: Optional[OFFAuthentication] = parse_auth(req)
+        auth: OFFAuthentication | None = parse_auth(req)
 
-        keep_types: Optional[list[str]] = req.get_param_as_list(
+        keep_types: list[str] | None = req.get_param_as_list(
             "insight_types", required=False
         )
         keep_types = filter_question_insight_types(keep_types)
@@ -1379,23 +1379,23 @@ def get_questions_resource_on_get(
     page: int = req.get_param_as_int("page", min_value=1, default=1)
     count: int = req.get_param_as_int("count", min_value=1, default=25)
     lang: str = req.get_param("lang", default="en")
-    keep_types: Optional[list[str]] = req.get_param_as_list(
+    keep_types: list[str] | None = req.get_param_as_list(
         "insight_types", required=False
     )
     keep_types = filter_question_insight_types(keep_types)
     value_tag: str = req.get_param("value_tag")
     brands = req.get_param_as_list("brands") or None
-    reserved_barcode: Optional[bool] = req.get_param_as_bool(
+    reserved_barcode: bool | None = req.get_param_as_bool(
         "reserved_barcode", default=False
     )
     server_type = get_server_type_from_req(req)
     countries = get_countries_from_req(req)
     # filter by annotation campaigns
-    campaigns: Optional[list[str]] = req.get_param_as_list("campaigns") or None
+    campaigns: list[str] | None = req.get_param_as_list("campaigns") or None
 
     if campaigns is None:
         # `campaign` is a deprecated field, use campaigns now instead
-        campaign: Optional[str] = req.get_param("campaign")
+        campaign: str | None = req.get_param("campaign")
         campaigns = [campaign] if campaign is not None else None
 
     predictor = req.get_param("predictor")
@@ -1404,7 +1404,7 @@ def get_questions_resource_on_get(
     # hash of the IPs as a backup.
     device_id = device_id_from_request(req)
 
-    auth: Optional[OFFAuthentication] = parse_auth(req)
+    auth: OFFAuthentication | None = parse_auth(req)
 
     if reserved_barcode:
         # Include all results, including non reserved barcodes
@@ -1542,10 +1542,10 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
         response: JSONType = {}
         count: int = req.get_param_as_int("count", min_value=1, default=25)
         page: int = req.get_param_as_int("page", min_value=1, default=1)
-        with_predictions: Optional[bool] = req.get_param_as_bool(
+        with_predictions: bool | None = req.get_param_as_bool(
             "with_predictions", default=False
         )
-        barcode: Optional[str] = normalize_req_barcode(req.get_param("barcode"))
+        barcode: str | None = normalize_req_barcode(req.get_param("barcode"))
         server_type = get_server_type_from_req(req)
 
         get_images_ = functools.partial(
@@ -1573,9 +1573,9 @@ class PredictionCollection:
     def on_get(self, req: falcon.Request, resp: falcon.Response):
         page: int = req.get_param_as_int("page", min_value=1, default=1)
         count: int = req.get_param_as_int("count", min_value=1, default=25)
-        barcode: Optional[str] = normalize_req_barcode(req.get_param("barcode"))
+        barcode: str | None = normalize_req_barcode(req.get_param("barcode"))
         value_tag: str = req.get_param("value_tag")
-        keep_types: Optional[list[str]] = req.get_param_as_list("types", required=False)
+        keep_types: list[str] | None = req.get_param_as_list("types", required=False)
         server_type = get_server_type_from_req(req)
 
         if keep_types:
@@ -1614,15 +1614,15 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
         count: int = req.get_param_as_int("count", min_value=1, default=25)
         insight_type: str = req.get_param("type")
         countries = get_countries_from_req(req)
-        reserved_barcode: Optional[bool] = req.get_param_as_bool(
+        reserved_barcode: bool | None = req.get_param_as_bool(
             "reserved_barcode", default=False
         )
         server_type = get_server_type_from_req(req)
         # filter by annotation campaigns
-        campaigns: Optional[list[str]] = req.get_param_as_list("campaigns") or None
+        campaigns: list[str] | None = req.get_param_as_list("campaigns") or None
         if campaigns is None:
             # `campaign` is a deprecated field, use campaigns now instead
-            campaign: Optional[str] = req.get_param("campaign")
+            campaign: str | None = req.get_param("campaign")
             campaigns = [campaign] if campaign is not None else None
 
         predictor = req.get_param("predictor")
@@ -1660,9 +1660,9 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
         response: JSONType = {}
         count: int = req.get_param_as_int("count", min_value=1, default=25)
         page: int = req.get_param_as_int("page", min_value=1, default=1)
-        with_logo: Optional[bool] = req.get_param_as_bool("with_logo", default=False)
-        barcode: Optional[str] = normalize_req_barcode(req.get_param("barcode"))
-        type: Optional[str] = req.get_param("type")
+        with_logo: bool | None = req.get_param_as_bool("with_logo", default=False)
+        barcode: str | None = normalize_req_barcode(req.get_param("barcode"))
+        type: str | None = req.get_param("type")
         server_type = get_server_type_from_req(req)
 
         query_parameters = {
@@ -1695,9 +1695,9 @@ def on_get(self, req: falcon.Request, resp: falcon.Response):
 class LogoAnnotationCollection:
     def on_get(self, req: falcon.Request, resp: falcon.Response):
         response: JSONType = {}
-        barcode: Optional[str] = normalize_req_barcode(req.get_param("barcode"))
+        barcode: str | None = normalize_req_barcode(req.get_param("barcode"))
         server_type = get_server_type_from_req(req)
-        keep_types: Optional[list[str]] = req.get_param_as_list("types", required=False)
+        keep_types: list[str] | None = req.get_param_as_list("types", required=False)
         value_tag: str = req.get_param("value_tag")
         page: int = req.get_param_as_int("page", min_value=1, default=1)
         count: int = req.get_param_as_int("count", min_value=1, default=25)

robotoff/app/core.py

@@ -1,7 +1,7 @@
 import datetime
 import functools
 from enum import Enum
-from typing import Iterable, Literal, NamedTuple, Optional, Union
+from typing import Iterable, Literal, NamedTuple, Union
 
 import falcon
 import peewee
@@ -65,27 +65,27 @@ def _add_vote_exclusion_clause(exclusion: SkipVotedOn) -> peewee.Expression:
 
 
 def get_insights(
-    barcode: Optional[str] = None,
+    barcode: str | None = None,
     server_type: ServerType = ServerType.off,
-    keep_types: Optional[list[str]] = None,
-    countries: Optional[list[Country]] = None,
-    brands: Optional[list[str]] = None,
-    annotated: Optional[bool] = False,
-    annotation: Optional[int] = None,
-    order_by: Optional[Literal["random", "popularity", "n_votes", "confidence"]] = None,
-    value_tag: Optional[str] = None,
-    reserved_barcode: Optional[bool] = None,
+    keep_types: list[str] | None = None,
+    countries: list[Country] | None = None,
+    brands: list[str] | None = None,
+    annotated: bool | None = False,
+    annotation: int | None = None,
+    order_by: Literal["random", "popularity", "n_votes", "confidence"] | None = None,
+    value_tag: str | None = None,
+    reserved_barcode: bool | None = None,
     as_dict: bool = False,
-    limit: Optional[int] = 25,
-    offset: Optional[int] = None,
+    limit: int | None = 25,
+    offset: int | None = None,
     count: bool = False,
-    max_count: Optional[int] = None,
-    avoid_voted_on: Optional[SkipVotedOn] = None,
-    group_by_value_tag: Optional[bool] = False,
-    automatically_processable: Optional[bool] = None,
-    campaigns: Optional[list[str]] = None,
-    predictor: Optional[str] = None,
-    lc: Optional[list[str]] = None,
+    max_count: int | None = None,
+    avoid_voted_on: SkipVotedOn | None = None,
+    group_by_value_tag: bool | None = False,
+    automatically_processable: bool | None = None,
+    campaigns: list[str] | None = None,
+    predictor: str | None = None,
+    lc: list[str] | None = None,
 ) -> Iterable[ProductInsight]:
     """Fetch insights that meet the criteria passed as parameters.
 
@@ -237,11 +237,11 @@ def get_insights(
 
 def get_images(
     server_type: ServerType,
-    with_predictions: Optional[bool] = False,
-    barcode: Optional[str] = None,
-    offset: Optional[int] = None,
+    with_predictions: bool | None = False,
+    barcode: str | None = None,
+    offset: int | None = None,
     count: bool = False,
-    limit: Optional[int] = None,
+    limit: int | None = None,
 ) -> Iterable[ImageModel]:
     where_clauses = [ImageModel.server_type == server_type.name]
 
@@ -273,11 +273,11 @@ def get_images(
 
 def get_predictions(
     server_type: ServerType,
-    barcode: Optional[str] = None,
-    keep_types: Optional[list[str]] = None,
-    value_tag: Optional[str] = None,
-    limit: Optional[int] = None,
-    offset: Optional[int] = None,
+    barcode: str | None = None,
+    keep_types: list[str] | None = None,
+    value_tag: str | None = None,
+    limit: int | None = None,
+    offset: int | None = None,
     count: bool = False,
 ) -> Iterable[Prediction]:
     where_clauses = [Prediction.server_type == server_type.name]
@@ -374,8 +374,8 @@ def save_annotation(
     annotation: InsightAnnotation,
     device_id: str,
     update: bool = True,
-    data: Optional[dict] = None,
-    auth: Optional[OFFAuthentication] = None,
+    data: dict | None = None,
+    auth: OFFAuthentication | None = None,
     trusted_annotator: bool = False,
 ) -> AnnotationResult:
     """Saves annotation either by using a single response as ground truth or
@@ -493,11 +493,11 @@ def save_annotation(
 
 def get_logo_annotation(
     server_type: ServerType,
-    barcode: Optional[str] = None,
-    keep_types: Optional[list[str]] = None,
-    value_tag: Optional[str] = None,
-    limit: Optional[int] = 25,
-    offset: Optional[int] = None,
+    barcode: str | None = None,
+    keep_types: list[str] | None = None,
+    value_tag: str | None = None,
+    limit: int | None = 25,
+    offset: int | None = None,
     count: bool = False,
 ) -> Iterable[LogoAnnotation]:
     """Return logos that fit the criteria passed as parameters.
@@ -550,7 +550,7 @@ def get_logo_annotation(
 
 
 def update_logo_annotations(
-    annotation_logos: list[tuple[str, Optional[str], LogoAnnotation]],
+    annotation_logos: list[tuple[str, str | None, LogoAnnotation]],
     username: str,
     completed_at: datetime.datetime,
 ) -> list[LogoAnnotation]:
@@ -582,7 +582,7 @@ def update_logo_annotations(
     return updated_logos
 
 
-def filter_question_insight_types(keep_types: Optional[list[str]]):
+def filter_question_insight_types(keep_types: list[str] | None):
     """Utility function to validate `insight_types` parameters in /questions/*
     queries.
 

robotoff/app/events.py

@@ -1,5 +1,4 @@
 from multiprocessing import Process, SimpleQueue
-from typing import Optional
 
 import requests
 
@@ -49,8 +48,8 @@ def send_event(
     event_type: str,
     user_id: str,
     device_id: str,
-    barcode: Optional[str] = None,
-    server_type: Optional[str] = None,
+    barcode: str | None = None,
+    server_type: str | None = None,
 ):
     event = {
         "event_type": event_type,

robotoff/batch/launch.py

@@ -1,7 +1,7 @@
 import datetime
 import re
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List
 
 import yaml
 from google.cloud import batch_v1
@@ -26,7 +26,7 @@ class GoogleBatchJobConfig(BaseModel):
     container_image_uri: str = Field(
         description="The URI of the container image to use for the job. SHould be a valid Image URI.",
     )
-    entrypoint: Optional[str] = Field(
+    entrypoint: str | None = Field(
         default=None,
         description="The entrypoint for the container. If None, use default entrypoint.",
         examples=["python main.py"],
@@ -46,7 +46,7 @@ class GoogleBatchJobConfig(BaseModel):
         description="The amount of RAM in MiB to allocate to each CPU core.",
         le=64000,
     )
-    boot_disk_mib: Optional[int] = Field(
+    boot_disk_mib: int | None = Field(
         default=None,
         description="The size of the boot disk in MiB. It is deleted once the job finished. If None, no bootDisk is added.",
         le=200000,  # 200 GB