chore(typing): modernizes codebase to use Python 3.9+ built-in types

src/dispatch/auth/models.py

@@ -7,7 +7,7 @@
 
 import bcrypt
 from jose import jwt
-from pydantic import field_validator, Field
+from pydantic import field_validator
 from pydantic import EmailStr
 
 from sqlalchemy import DateTime, Column, String, LargeBinary, Integer, Boolean
@@ -131,14 +131,14 @@ class UserProject(DispatchBase):
     """Pydantic model for a user's project membership."""
     project: ProjectRead
     default: bool | None = False
-    role: str | None = Field(None, nullable=True)
+    role: str | None = None
 
 
 class UserOrganization(DispatchBase):
     """Pydantic model for a user's organization membership."""
     organization: OrganizationRead
     default: bool | None = False
-    role: str | None = Field(None, nullable=True)
+    role: str | None = None
 
 
 class UserBase(DispatchBase):
@@ -171,7 +171,7 @@ def password_required(cls, v):
 
 class UserRegister(UserLogin):
     """Pydantic model for user registration data."""
-    password: str = Field(None, nullable=True)
+    password: str = None
 
     @field_validator("password", mode="before")
     @classmethod
@@ -184,13 +184,13 @@ def password_required(cls, v):
 class UserLoginResponse(DispatchBase):
     """Pydantic model for the response after user login."""
     projects: list[UserProject] | None
-    token: str | None = Field(None, nullable=True)
+    token: str | None = None
 
 
 class UserRead(UserBase):
     """Pydantic model for reading user data."""
     id: PrimaryKey
-    role: str | None = Field(None, nullable=True)
+    role: str | None = None
     experimental_features: bool | None
 
 
@@ -200,7 +200,7 @@ class UserUpdate(DispatchBase):
     projects: list[UserProject] | None
     organizations: list[UserOrganization] | None
     experimental_features: bool | None
-    role: str | None = Field(None, nullable=True)
+    role: str | None = None
 
 
 class UserPasswordUpdate(DispatchBase):
@@ -249,10 +249,10 @@ def validate_password(cls, v):
 class UserCreate(DispatchBase):
     """Pydantic model for creating a new user."""
     email: EmailStr
-    password: str | None = Field(None, nullable=True)
+    password: str | None = None
     projects: list[UserProject] | None
     organizations: list[UserOrganization] | None
-    role: str | None = Field(None, nullable=True)
+    role: str | None = None
 
     @field_validator("password", mode="before")
     @classmethod
@@ -263,7 +263,7 @@ def hash(cls, v):
 
 class UserRegisterResponse(DispatchBase):
     """Pydantic model for the response after user registration."""
-    token: str | None = Field(None, nullable=True)
+    token: str | None = None
 
 
 class UserPagination(Pagination):

src/dispatch/auth/service.py

@@ -6,7 +6,7 @@
 """
 
 import logging
-from typing import Annotated, Optional
+from typing import Annotated
 
 from fastapi import HTTPException, Depends
 from starlette.requests import Request
@@ -44,12 +44,12 @@
 )
 
 
-def get(*, db_session, user_id: int) -> Optional[DispatchUser]:
+def get(*, db_session, user_id: int) -> DispatchUser | None:
     """Returns a user based on the given user id."""
     return db_session.query(DispatchUser).filter(DispatchUser.id == user_id).one_or_none()
 
 
-def get_by_email(*, db_session, email: str) -> Optional[DispatchUser]:
+def get_by_email(*, db_session, email: str) -> DispatchUser | None:
     """Returns a user object based on user email."""
     return db_session.query(DispatchUser).filter(DispatchUser.email == email).one_or_none()
 

src/dispatch/case/flows.py

@@ -1,6 +1,5 @@
 import logging
 from datetime import datetime
-from typing import List, Optional
 
 from sqlalchemy.orm import Session
 
@@ -742,8 +741,8 @@ def send_escalation_messages_for_channel_case(
 
 
 def map_case_roles_to_incident_roles(
-    participant_roles: List[ParticipantRole], incident: Incident, db_session: Session
-) -> Optional[List[ParticipantRoleType]]:
+    participant_roles: list[ParticipantRole], incident: Incident, db_session: Session
+) -> list[ParticipantRoleType | None]:
     # Map the case role to an incident role
     incident_roles = set()
     for role in participant_roles:
@@ -1001,8 +1000,8 @@ def case_create_conversation_flow(
 def case_create_resources_flow(
     db_session: Session,
     case_id: int,
-    individual_participants: List[str],
-    team_participants: List[str],
+    individual_participants: list[str],
+    team_participants: list[str],
     conversation_target: str = None,
     create_all_resources: bool = True,
 ) -> None:

src/dispatch/case/messaging.py

@@ -7,7 +7,6 @@
 
 import logging
 
-from typing import Optional
 
 from sqlalchemy.orm import Session
 
@@ -314,7 +313,7 @@ def send_case_welcome_participant_message(
     participant_email: str,
     case: Case,
     db_session: Session,
-    welcome_template: Optional[EmailTemplates] = None,
+    welcome_template: EmailTemplates | None = None,
 ):
     if not case.dedicated_channel:
         return

src/dispatch/case/models.py

@@ -2,7 +2,6 @@
 from collections import Counter, defaultdict
 from datetime import datetime
 from typing import Any
-
 from pydantic import field_validator, Field
 from sqlalchemy import (
     Boolean,

src/dispatch/case/priority/models.py

@@ -1,6 +1,4 @@
 """Models and schemas for the Dispatch case priority system."""
-from pydantic import Field
-
 from sqlalchemy import Column, Integer, String, Boolean
 from sqlalchemy.sql.schema import UniqueConstraint
 from sqlalchemy.event import listen
@@ -37,10 +35,10 @@ class CasePriority(Base, ProjectMixin):
 # Pydantic models
 class CasePriorityBase(DispatchBase):
     """Base Pydantic model for case priority data."""
-    color: str | None = Field(None, nullable=True)
+    color: str | None = None
     default: bool | None
     page_assignee: bool | None
-    description: str | None = Field(None, nullable=True)
+    description: str | None = None
     enabled: bool | None
     name: NameStr
     project: ProjectRead | None

src/dispatch/case/priority/service.py

@@ -1,4 +1,3 @@
-from typing import List, Optional
 from pydantic import ValidationError
 
 from sqlalchemy.sql.expression import true
@@ -13,7 +12,7 @@
 )
 
 
-def get(*, db_session, case_priority_id: int) -> Optional[CasePriority]:
+def get(*, db_session, case_priority_id: int) -> CasePriority | None:
     """Returns a case priority based on the given priority id."""
     return db_session.query(CasePriority).filter(CasePriority.id == case_priority_id).one_or_none()
 
@@ -47,7 +46,7 @@ def get_default_or_raise(*, db_session, project_id: int) -> CasePriority:
     return case_priority
 
 
-def get_by_name(*, db_session, project_id: int, name: str) -> Optional[CasePriority]:
+def get_by_name(*, db_session, project_id: int, name: str) -> CasePriority | None:
     """Returns a case priority based on the given priority name."""
     return (
         db_session.query(CasePriority)
@@ -96,14 +95,14 @@ def get_by_name_or_default(
     return get_default_or_raise(db_session=db_session, project_id=project_id)
 
 
-def get_all(*, db_session, project_id: int = None) -> List[Optional[CasePriority]]:
+def get_all(*, db_session, project_id: int = None) -> list[CasePriority | None]:
     """Returns all case priorities."""
     if project_id is not None:
         return db_session.query(CasePriority).filter(CasePriority.project_id == project_id)
     return db_session.query(CasePriority)
 
 
-def get_all_enabled(*, db_session, project_id: int = None) -> List[Optional[CasePriority]]:
+def get_all_enabled(*, db_session, project_id: int = None) -> list[CasePriority | None]:
     """Returns all enabled case priorities."""
     if project_id is not None:
         return (

src/dispatch/case/service.py

@@ -4,7 +4,6 @@
 
 from pydantic import ValidationError
 from sqlalchemy.orm import Session, joinedload, load_only
-from typing import List, Optional
 
 from dispatch.auth.models import DispatchUser
 from dispatch.case.priority import service as case_priority_service
@@ -32,12 +31,12 @@
 log = logging.getLogger(__name__)
 
 
-def get(*, db_session, case_id: int) -> Optional[Case]:
+def get(*, db_session, case_id: int) -> Case | None:
     """Returns a case based on the given id."""
     return db_session.query(Case).filter(Case.id == case_id).first()
 
 
-def get_by_name(*, db_session, project_id: int, name: str) -> Optional[Case]:
+def get_by_name(*, db_session, project_id: int, name: str) -> Case | None:
     """Returns a case based on the given name."""
     return (
         db_session.query(Case)
@@ -64,12 +63,12 @@ def get_by_name_or_raise(*, db_session, project_id: int, case_in: CaseRead) -> C
     return case
 
 
-def get_all(*, db_session, project_id: int) -> List[Optional[Case]]:
+def get_all(*, db_session, project_id: int) -> list[Case | None]:
     """Returns all cases."""
     return db_session.query(Case).filter(Case.project_id == project_id)
 
 
-def get_all_open_by_case_type(*, db_session, case_type_id: int) -> List[Optional[Case]]:
+def get_all_open_by_case_type(*, db_session, case_type_id: int) -> list[Case | None]:
     """Returns all non-closed cases based on the given case type."""
     return (
         db_session.query(Case)
@@ -82,7 +81,7 @@ def get_all_open_by_case_type(*, db_session, case_type_id: int) -> List[Optional
 
 def get_all_by_status(
     *, db_session: Session, project_id: int, statuses: list[str]
-) -> List[Optional[Case]]:
+) -> list[Case | None]:
     """Returns all cases based on a given list of statuses."""
     return (
         db_session.query(Case)
@@ -103,15 +102,15 @@ def get_all_by_status(
     )
 
 
-def get_all_last_x_hours(*, db_session, hours: int) -> List[Optional[Case]]:
+def get_all_last_x_hours(*, db_session, hours: int) -> list[Case | None]:
     """Returns all cases in the last x hours."""
     now = datetime.utcnow()
     return db_session.query(Case).filter(Case.created_at >= now - timedelta(hours=hours)).all()
 
 
 def get_all_last_x_hours_by_status(
     *, db_session, project_id: int, status: str, hours: int
-) -> List[Optional[Case]]:
+) -> list[Case | None]:
     """Returns all cases of a given status in the last x hours."""
     now = datetime.utcnow()
 

src/dispatch/case/severity/models.py

@@ -1,7 +1,5 @@
 """Models and schemas for the Dispatch case severity system."""
 
-from pydantic import Field
-
 from sqlalchemy import Column, Integer, String, Boolean
 from sqlalchemy.sql.schema import UniqueConstraint
 from sqlalchemy.event import listen
@@ -41,9 +39,9 @@ class CaseSeverity(Base, ProjectMixin):
 # Pydantic models
 class CaseSeverityBase(DispatchBase):
     """Base Pydantic model for case severity data."""
-    color: str | None = Field(None, nullable=True)
+    color: str | None = None
     default: bool | None
-    description: str | None = Field(None, nullable=True)
+    description: str | None = None
     enabled: bool | None
     name: NameStr
     project: ProjectRead | None

src/dispatch/case/severity/service.py

@@ -1,4 +1,3 @@
-from typing import List, Optional
 from pydantic import ValidationError
 
 from sqlalchemy.sql.expression import true
@@ -13,7 +12,7 @@
 )
 
 
-def get(*, db_session, case_severity_id: int) -> Optional[CaseSeverity]:
+def get(*, db_session, case_severity_id: int) -> CaseSeverity | None:
     """Returns a case severity based on the given severity id."""
     return db_session.query(CaseSeverity).filter(CaseSeverity.id == case_severity_id).one_or_none()
 
@@ -43,7 +42,7 @@ def get_default_or_raise(*, db_session, project_id: int) -> CaseSeverity:
     return case_severity
 
 
-def get_by_name(*, db_session, project_id: int, name: str) -> Optional[CaseSeverity]:
+def get_by_name(*, db_session, project_id: int, name: str) -> CaseSeverity | None:
     """Returns a case severity based on the given severity name."""
     return (
         db_session.query(CaseSeverity)
@@ -88,14 +87,14 @@ def get_by_name_or_default(
     return get_default_or_raise(db_session=db_session, project_id=project_id)
 
 
-def get_all(*, db_session, project_id: int = None) -> List[Optional[CaseSeverity]]:
+def get_all(*, db_session, project_id: int = None) -> list[CaseSeverity | None]:
     """Returns all case severities."""
     if project_id:
         return db_session.query(CaseSeverity).filter(CaseSeverity.project_id == project_id)
     return db_session.query(CaseSeverity)
 
 
-def get_all_enabled(*, db_session, project_id: int = None) -> List[Optional[CaseSeverity]]:
+def get_all_enabled(*, db_session, project_id: int = None) -> list[CaseSeverity | None]:
     """Returns all enabled case severities."""
     if project_id:
         return (

src/dispatch/case/type/service.py

@@ -1,4 +1,3 @@
-from typing import List, Optional
 
 from sqlalchemy.sql.expression import true
 
@@ -13,7 +12,7 @@
 from .models import CaseType, CaseTypeCreate, CaseTypeRead, CaseTypeUpdate
 
 
-def get(*, db_session, case_type_id: int) -> Optional[CaseType]:
+def get(*, db_session, case_type_id: int) -> CaseType | None:
     """Returns a case type based on the given type id."""
     return db_session.query(CaseType).filter(CaseType.id == case_type_id).one_or_none()
 
@@ -37,7 +36,7 @@ def get_default_or_raise(*, db_session, project_id: int) -> CaseType:
     return case_type
 
 
-def get_by_name(*, db_session, project_id: int, name: str) -> Optional[CaseType]:
+def get_by_name(*, db_session, project_id: int, name: str) -> CaseType | None:
     """Returns a case type based on the given type name."""
     return (
         db_session.query(CaseType)
@@ -67,7 +66,7 @@ def get_by_name_or_default(*, db_session, project_id: int, case_type_in=CaseType
     return get_default_or_raise(db_session=db_session, project_id=project_id)
 
 
-def get_by_slug(*, db_session, project_id: int, slug: str) -> Optional[CaseType]:
+def get_by_slug(*, db_session, project_id: int, slug: str) -> CaseType | None:
     """Returns a case type based on the given type slug."""
     return (
         db_session.query(CaseType)
@@ -77,14 +76,14 @@ def get_by_slug(*, db_session, project_id: int, slug: str) -> Optional[CaseType]
     )
 
 
-def get_all(*, db_session, project_id: int = None) -> List[Optional[CaseType]]:
+def get_all(*, db_session, project_id: int = None) -> list[CaseType | None]:
     """Returns all case types."""
     if project_id:
         return db_session.query(CaseType).filter(CaseType.project_id == project_id)
     return db_session.query(CaseType)
 
 
-def get_all_enabled(*, db_session, project_id: int = None) -> List[Optional[CaseType]]:
+def get_all_enabled(*, db_session, project_id: int = None) -> list[CaseType | None]:
     """Returns all enabled case types."""
     if project_id:
         return (