Small security flags and linter feedback

alembic_osm/env.py

@@ -8,11 +8,11 @@
 # Add the project root directory to the Python path
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
+from alembic import context
 from sqlalchemy import pool
 from sqlalchemy.engine import Connection
 from sqlalchemy.ext.asyncio import async_engine_from_config
 
-from alembic import context
 from api.core.config import settings
 from api.core.database import Base
 

alembic_osm/versions/9221408912dd_add_user_role_table.py

@@ -5,68 +5,84 @@
 Create Date: 2026-01-29 14:54:10.669000
 
 """
+
 from typing import Sequence, Union
 
+import sqlalchemy as sa
 from alembic import op
 from sqlalchemy import inspect, text
-import sqlalchemy as sa
-
 
 # revision identifiers, used by Alembic.
-revision: str = '9221408912dd'
+revision: str = "9221408912dd"
 down_revision: Union[str, None] = None
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
     bind = op.get_bind()
+    assert bind is not None
     insp = inspect(bind)
 
     # Add unique constraint on users.auth_uid (if not already present)
     constraint_exists = bind.execute(
         text("SELECT 1 FROM pg_constraint WHERE conname = 'auth_uid_unique'")
     ).scalar()
     if not constraint_exists:
-        op.create_unique_constraint('auth_uid_unique', 'users', ['auth_uid'])
+        op.create_unique_constraint("auth_uid_unique", "users", ["auth_uid"])
 
     # Create the workspace_role enum type (if not already present)
     result = bind.execute(
         text("SELECT 1 FROM pg_type WHERE typname = 'workspace_role'")
     )
     if not result.scalar():
-        workspace_role = sa.Enum('lead', 'validator', 'contributor', name='workspace_role')
+        workspace_role = sa.Enum(
+            "lead", "validator", "contributor", name="workspace_role"
+        )
         workspace_role.create(bind)
 
     # Create the user_workspace_roles table (if not already present)
-    if not insp.has_table('user_workspace_roles'):
+    if not insp.has_table("user_workspace_roles"):
         op.create_table(
-            'user_workspace_roles',
-            sa.Column('user_auth_uid', sa.Uuid(), nullable=False),
-            sa.Column('workspace_id', sa.BigInteger(), nullable=False),
-            sa.Column('role', sa.Enum('lead', 'validator', 'contributor', name='workspace_role', create_type=False), nullable=False),
-            sa.ForeignKeyConstraint(['user_auth_uid'], ['users.auth_uid']),
-            sa.PrimaryKeyConstraint('user_auth_uid', 'workspace_id')
+            "user_workspace_roles",
+            sa.Column("user_auth_uid", sa.Uuid(), nullable=False),
+            sa.Column("workspace_id", sa.BigInteger(), nullable=False),
+            sa.Column(
+                "role",
+                sa.Enum(
+                    "lead",
+                    "validator",
+                    "contributor",
+                    name="workspace_role",
+                    create_type=False,
+                ),
+                nullable=False,
+            ),
+            sa.ForeignKeyConstraint(["user_auth_uid"], ["users.auth_uid"]),
+            sa.PrimaryKeyConstraint("user_auth_uid", "workspace_id"),
         )
 
 
 def downgrade() -> None:
     bind = op.get_bind()
+    assert bind is not None
     insp = inspect(bind)
 
-    if insp.has_table('user_workspace_roles'):
-        op.drop_table('user_workspace_roles')
+    if insp.has_table("user_workspace_roles"):
+        op.drop_table("user_workspace_roles")
 
     # Drop the enum type
     result = bind.execute(
         text("SELECT 1 FROM pg_type WHERE typname = 'workspace_role'")
     )
     if result.scalar():
-        workspace_role = sa.Enum('lead', 'validator', 'contributor', name='workspace_role')
+        workspace_role = sa.Enum(
+            "lead", "validator", "contributor", name="workspace_role"
+        )
         workspace_role.drop(bind)
 
     constraint_exists = bind.execute(
         text("SELECT 1 FROM pg_constraint WHERE conname = 'auth_uid_unique'")
     ).scalar()
     if constraint_exists:
-        op.drop_constraint('auth_uid_unique', 'users', type_='unique')
+        op.drop_constraint("auth_uid_unique", "users", type_="unique")

alembic_task/env.py

@@ -8,11 +8,11 @@
 # Add the project root directory to the Python path
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
+from alembic import context
 from sqlalchemy import pool
 from sqlalchemy.engine import Connection
 from sqlalchemy.ext.asyncio import async_engine_from_config
 
-from alembic import context
 from api.core.config import settings
 from api.core.database import Base
 

alembic_task/versions/add6266277c7_.py

@@ -7,7 +7,6 @@
 """
 
 import sqlalchemy as sa
-
 from alembic import op
 
 # revision identifiers, used by Alembic.

api/core/config.py

@@ -12,8 +12,12 @@ class Settings(BaseSettings):
     #
     CORS_ORIGINS: list[str] = []
 
-    TASK_DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/tasking_manager"
-    OSM_DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/tasking_manager"
+    TASK_DATABASE_URL: str = (
+        "postgresql+asyncpg://user:pass@localhost:5432/tasking_manager"
+    )
+    OSM_DATABASE_URL: str = (
+        "postgresql+asyncpg://user:pass@localhost:5432/tasking_manager"
+    )
 
     TDEI_BACKEND_URL: str = "https://portal-api-dev.tdei.us/api/v1/"
     TDEI_OIDC_URL: str = "https://account-dev.tdei.us/"
@@ -28,7 +32,6 @@ class Settings(BaseSettings):
 
     # proxy destination--"osm-web" is a virtual docker network endpoint
     WS_OSM_HOST: str = "http://osm-web"
-    #WS_OSM_HOST: str = "https://osm.workspaces-dev.sidewalks.washington.edu"
 
     SENTRY_DSN: str = ""
 
@@ -37,4 +40,5 @@ class Settings(BaseSettings):
         env_file_encoding="utf-8",
     )
 
+
 settings = Settings()

api/main.py

@@ -1,5 +1,6 @@
 import os
 import re
+import sys
 from contextlib import asynccontextmanager
 
 import httpx
@@ -36,9 +37,6 @@
 # Set up logging configuration
 setup_logging()
 
-# Optional: Run migrations on startup
-run_migrations()
-
 # Set up logger for this module
 logger = get_logger(__name__)
 
@@ -48,6 +46,10 @@
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI):
+    # only run migrations when not under test
+    if "pytest" not in sys.modules:
+        run_migrations()
+
     # Run before app bootstrap:
     global _osm_client
     _osm_client = httpx.AsyncClient(

api/src/teams/repository.py

@@ -24,7 +24,7 @@ async def get_all(self, workspace_id: int) -> list[WorkspaceTeamItem]:
             .where(WorkspaceTeam.workspace_id == workspace_id)
         )
 
-        return [WorkspaceTeamItem.from_team(x) for x in result.scalars()]
+        return [WorkspaceTeamItem.from_team(x) for x in result.scalars().all()]
 
     async def get(self, id: int, load_members: bool = False) -> WorkspaceTeam:
         query = select(WorkspaceTeam).where(WorkspaceTeam.id == id)
@@ -59,14 +59,13 @@ async def assert_team_in_workspace(self, id: int, workspace_id: int):
             raise NotFoundException(f"Team {id} not in workspace {workspace_id}")
 
     async def create(self, workspace_id: int, data: WorkspaceTeamCreate) -> int:
-        team = WorkspaceTeam()
-        team.workspace_id = workspace_id
-        team.name = data.name
+        team = WorkspaceTeam(name=data.name, workspace_id=workspace_id)
 
         self.session.add(team)
         await self.session.commit()
         await self.session.refresh(team)
 
+        assert team.id is not None
         return team.id
 
     async def update(self, id: int, data: WorkspaceTeamUpdate):
@@ -77,7 +76,7 @@ async def update(self, id: int, data: WorkspaceTeamUpdate):
         await self.session.commit()
 
     async def delete(self, id: int) -> None:
-        await self.session.exec(delete(WorkspaceTeam).where(WorkspaceTeam.id == id))
+        await self.session.execute(delete(WorkspaceTeam).where(WorkspaceTeam.id == id))  # type: ignore[arg-type]
         await self.session.commit()
 
     async def get_members(self, id: int) -> list[User]:

api/src/teams/routes.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, status
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from api.core.database import get_osm_session, get_task_session

api/src/teams/schemas.py

@@ -1,4 +1,4 @@
-from typing import Self, TYPE_CHECKING
+from typing import TYPE_CHECKING, Self
 
 from sqlmodel import Field, Relationship, SQLModel
 

api/src/workspaces/repository.py

@@ -83,7 +83,7 @@ async def update(
         )
         result = await self.session.execute(query)
 
-        if result.rowcount != 1:
+        if result.rowcount != 1:  # type: ignore[attr-defined]
             raise NotFoundException(f"Update failed for workspace id {workspace_id}")
 
         await self.session.commit()
@@ -134,7 +134,7 @@ async def updateLongformQuest(
         )
         result = await self.session.execute(query)
 
-        if result.rowcount == 0:
+        if result.rowcount == 0:  # type: ignore[attr-defined]
             raise NotFoundException(f"Workspace with id {workspace_id} not found")
 
         await self.session.commit()
@@ -181,7 +181,7 @@ async def updateImageryDef(
 
         result = await self.session.execute(query)
 
-        if result.rowcount != 1:
+        if result.rowcount != 1:  # type: ignore[attr-defined]
             raise NotFoundException(f"Update failed for workspace id {workspace_id}")
 
         await self.session.commit()
@@ -195,7 +195,7 @@ async def delete(self, current_user: UserInfo, workspace_id: int) -> None:
 
         result = await self.session.execute(query)
 
-        if result.rowcount != 1:
+        if result.rowcount != 1:  # type: ignore[attr-defined]
             raise NotFoundException(f"Workspace delete failed for id {workspace_id}")
 
         await self.session.commit()
@@ -257,7 +257,7 @@ async def addUserToWorkspaceWithRole(
     ) -> None:
 
         userRole = WorkspaceUserRole(
-            auth_user_uid=cast(UUID, current_user.user_uuid),
+            auth_user_uid=cast(UUID, user_id),
             workspace_id=workspace_id,
             role=role,
         )
@@ -284,7 +284,7 @@ async def removeUserFromWorkspace(
 
         result = await self.session.execute(query)
 
-        if result.rowcount != 1:
+        if result.rowcount != 1:  # type: ignore[attr-defined]
             raise NotFoundException(
                 f"User association removal failed for workspace {workspace_id} and user {user_id}"
             )

api/src/workspaces/schemas.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from enum import IntEnum, StrEnum
-from typing import Any, Optional, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Optional
 from uuid import UUID
 
 from geoalchemy2 import Geometry