From d13ffc2319cd309667f2b479d50ebd01dd2558bf Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Wed, 12 Aug 2026 15:47:38 +0530 Subject: [PATCH 1/9] Added initial items --- Dockerfile | 3 + .../versions/37c12e8301ee_workspace_jobs.py | 56 +++++++++ api/main.py | 2 + api/src/workspaces/jobs/repository.py | 108 ++++++++++++++++++ api/src/workspaces/jobs/routes.py | 24 ++++ api/src/workspaces/jobs/schemas.py | 47 ++++++++ api/src/workspaces/routes.py | 24 ++++ docker-compose.local.yml | 2 +- 8 files changed, 265 insertions(+), 1 deletion(-) create mode 100644 alembic_osm/versions/37c12e8301ee_workspace_jobs.py create mode 100644 api/src/workspaces/jobs/repository.py create mode 100644 api/src/workspaces/jobs/routes.py create mode 100644 api/src/workspaces/jobs/schemas.py diff --git a/Dockerfile b/Dockerfile index 0ca05c9..8e35185 100644 --- a/Dockerfile +++ b/Dockerfile @@ -17,5 +17,8 @@ ADD . /app RUN --mount=type=cache,target=/root/.cache/uv \ uv sync --frozen +# set the environment variable for uvicorn to run the FastAPI app +ENV UV_PROJECT_ENVIRONMENT=.uvenv + # Run with uvicorn CMD ["uv", "run", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/alembic_osm/versions/37c12e8301ee_workspace_jobs.py b/alembic_osm/versions/37c12e8301ee_workspace_jobs.py new file mode 100644 index 0000000..a615c0f --- /dev/null +++ b/alembic_osm/versions/37c12e8301ee_workspace_jobs.py @@ -0,0 +1,56 @@ +"""workspace jobs + +Revision ID: 37c12e8301ee +Revises: a92361f527ef +Create Date: 2026-08-12 10:03:15.757771 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "37c12e8301ee" +down_revision: Union[str, None] = "a92361f527ef" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "jobs", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("job_type", sa.String(), nullable=False), + sa.Column("status", sa.String(), nullable=False), + sa.Column("request", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column( + "created_at", + sa.DateTime(), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column("current_task", sa.String(), nullable=True), + sa.Column("current_task_status", sa.String(), nullable=True), + sa.Column("response", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column("workspace_id", sa.Integer(), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + # ### end Alembic commands ### + pass + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("jobs") + # ### end Alembic commands ### + pass diff --git a/api/main.py b/api/main.py index c3708b3..65c6aee 100644 --- a/api/main.py +++ b/api/main.py @@ -30,6 +30,7 @@ from api.src.tasking.tasks.routes import router as tasking_tasks_router from api.src.teams.routes import router as teams_router from api.src.users.routes import router as users_router +from api.src.workspaces.jobs.routes import router as jobs_router from api.src.workspaces.repository import WorkspaceRepository from api.src.workspaces.routes import router as workspaces_router from api.utils.migrations import run_migrations @@ -108,6 +109,7 @@ async def lifespan(_app: FastAPI): app.include_router(osm_router, prefix="/api/v1") app.include_router(teams_router, prefix="/api/v1") app.include_router(users_router, prefix="/api/v1") +app.include_router(jobs_router, prefix="/api/v1") app.include_router(workspaces_router, prefix="/api/v1") app.include_router(tasking_projects_router, prefix="/api/v1") app.include_router(tasking_me_router, prefix="/api/v1") diff --git a/api/src/workspaces/jobs/repository.py b/api/src/workspaces/jobs/repository.py new file mode 100644 index 0000000..6388f95 --- /dev/null +++ b/api/src/workspaces/jobs/repository.py @@ -0,0 +1,108 @@ +from sqlalchemy import delete, select, update +from sqlalchemy.exc import IntegrityError +from sqlmodel.ext.asyncio.session import AsyncSession + +from api.core.exceptions import ( + AlreadyExistsException, + ForbiddenException, + NotFoundException, +) +from api.core.security import UserInfo +from api.src.workspaces.jobs.schemas import Job, JobCreate, JobPatch + + +class JobRepository: + + def __init__(self, session: AsyncSession): + self.session = session + + @staticmethod + def _accessible_workspace_ids(current_user: UserInfo) -> list[int]: + workspace_ids: list[int] = [] + for ids in current_user.accessibleWorkspaceIds.values(): + workspace_ids.extend(ids) + return workspace_ids + + async def create(self, current_user: UserInfo, job_data: JobCreate) -> Job: + if not current_user.isWorkspaceContributor(job_data.workspace_id): + raise ForbiddenException( + "User does not have permissions to create a job in that workspace." + ) + + job = Job(**job_data.model_dump()) + + try: + self.session.add(job) + await self.session.commit() + await self.session.refresh(job) + return job + except IntegrityError: + await self.session.rollback() + raise AlreadyExistsException(f"Job with ID {job.id} already exists") + + async def getById(self, current_user: UserInfo, job_id: int) -> Job: + accessible_workspace_ids = self._accessible_workspace_ids(current_user) + + query = select(Job).where( + (Job.id == job_id) + & (Job.workspace_id.in_(accessible_workspace_ids)) # type: ignore[attr-defined] + ) + result = await self.session.execute(query) + job = result.scalar_one_or_none() + + if not job: + raise NotFoundException(f"Job with id {job_id} not found") + + return job + + async def update( + self, + current_user: UserInfo, + job_id: int, + job_data: JobPatch, + ) -> Job: + accessible_workspace_ids = self._accessible_workspace_ids(current_user) + + query = ( + update(Job) + .where( + (Job.id == job_id) + & (Job.workspace_id.in_(accessible_workspace_ids)) # type: ignore[attr-defined] + ) + .values(**job_data.model_dump(exclude_unset=True)) + ) + + result = await self.session.execute(query) + + if result.rowcount != 1: # type: ignore[attr-defined] + raise NotFoundException(f"Update failed for job id {job_id}") + + await self.session.commit() + return await self.getById(current_user, job_id) + + async def delete(self, current_user: UserInfo, job_id: int) -> None: + accessible_workspace_ids = self._accessible_workspace_ids(current_user) + + query = delete(Job).where( + (Job.id == job_id) + & (Job.workspace_id.in_(accessible_workspace_ids)) # type: ignore[attr-defined] + ) + + result = await self.session.execute(query) + + if result.rowcount != 1: # type: ignore[attr-defined] + raise NotFoundException(f"Job delete failed for id {job_id}") + + await self.session.commit() + + async def getWorkspaceJobs( + self, current_user: UserInfo, workspace_id: int + ) -> list[Job]: + if not current_user.isWorkspaceContributor(workspace_id): + raise ForbiddenException( + "User does not have permissions to view jobs in that workspace." + ) + + query = select(Job).where(Job.workspace_id == workspace_id) + result = await self.session.execute(query) + return list(result.scalars().all()) diff --git a/api/src/workspaces/jobs/routes.py b/api/src/workspaces/jobs/routes.py new file mode 100644 index 0000000..72816d4 --- /dev/null +++ b/api/src/workspaces/jobs/routes.py @@ -0,0 +1,24 @@ +from fastapi import APIRouter, Depends +from sqlmodel.ext.asyncio.session import AsyncSession + +from api.core.database import get_osm_session +from api.core.security import UserInfo, validate_token +from api.src.workspaces.jobs.repository import JobRepository +from api.src.workspaces.jobs.schemas import Job + +router = APIRouter(prefix="/workspaces/jobs", tags=["jobs"]) + + +def get_job_repository( + session: AsyncSession = Depends(get_osm_session), +) -> JobRepository: + return JobRepository(session) + + +@router.get("/{job_id}", response_model=Job) +async def get_job_by_id( + job_id: int, + repository: JobRepository = Depends(get_job_repository), + current_user: UserInfo = Depends(validate_token), +) -> Job: + return await repository.getById(current_user, job_id) diff --git a/api/src/workspaces/jobs/schemas.py b/api/src/workspaces/jobs/schemas.py new file mode 100644 index 0000000..273e38b --- /dev/null +++ b/api/src/workspaces/jobs/schemas.py @@ -0,0 +1,47 @@ +from datetime import datetime +from typing import Any + +from sqlalchemy import JSON as SAJson +from sqlalchemy import Column +from sqlmodel import Field, Relationship, SQLModel + + +class Job(SQLModel, table=True): + __tablename__ = "jobs" + + id: int = Field(default=None, primary_key=True) + job_type: str = Field(default=None, nullable=False) + status: str = Field(default=None, nullable=False) + request: dict = Field(default=None, sa_column=Column(SAJson, nullable=False)) + created_at: datetime = Field(sa_column=Column(nullable=False, default=datetime.now)) + updated_at: datetime = Field(sa_column=Column(nullable=False, default=datetime.now)) + current_task: str = Field(default=None, nullable=True) + current_task_status: str = Field(default=None, nullable=True) + response: dict = Field(default=None, sa_column=Column(SAJson, nullable=True)) + workspace_id: int = Field(default=None) # Not sure if we have to add index here. + + # workspace_id: int = Field(foreign_key="workspaces.id") + # workspace: "Workspace" = Relationship(back_populates="jobs") + + +class JobCreate(SQLModel): + """Fields the client may supply when creating a job.""" + + job_type: str + status: str + request: dict[str, Any] + workspace_id: int + current_task: str | None = None + current_task_status: str | None = None + response: dict[str, Any] | None = None + + +class JobPatch(SQLModel): + """Fields the client may supply when updating a job.""" + + job_type: str | None = None + status: str | None = None + request: dict[str, Any] | None = None + current_task: str | None = None + current_task_status: str | None = None + response: dict[str, Any] | None = None diff --git a/api/src/workspaces/routes.py b/api/src/workspaces/routes.py index e13ea1d..ec25390 100644 --- a/api/src/workspaces/routes.py +++ b/api/src/workspaces/routes.py @@ -16,6 +16,8 @@ from api.src.tasking.projects.repository import TaskingProjectRepository from api.src.users.repository import UserRepository from api.src.users.schemas import WorkspaceUserRoleType +from api.src.workspaces.jobs.repository import JobRepository +from api.src.workspaces.jobs.schemas import Job from api.src.workspaces.repository import WorkspaceRepository from api.src.workspaces.schemas import ( ImagerySettingsPatch, @@ -53,6 +55,12 @@ def get_project_repository( return TaskingProjectRepository(session) +def get_jobs_repository( + session: AsyncSession = Depends(get_osm_session), +) -> JobRepository: + return JobRepository(session) + + # @test: Test that this endpoint properly handles any exceptions and returns a 500 if an unexpected error occurs # @test: Test that this method properly calls the repository method to fetch the workspace and that the repository method properly fetches the workspace from the database # @test: Test that this method properly handles numeric workspace_id input and invalid values for the same @@ -172,6 +180,7 @@ async def create_workspace( workspace_data: WorkspaceCreate, repository_ws: WorkspaceRepository = Depends(get_workspace_repository), repository_users: UserRepository = Depends(get_user_repository), + jobs_repository: JobRepository = Depends(get_jobs_repository), current_user: UserInfo = Depends(validate_token), ) -> dict[str, int]: try: @@ -187,6 +196,7 @@ async def create_workspace( WorkspaceUserRoleType.LEAD, ) + # await jobs_repository.create(current_user, workspace.id) # Evict the creator's cache so their next request reflects the new # workspace and lead role rather than serving stale data for up to # an hour: @@ -455,3 +465,17 @@ async def update_imagery_settings( except Exception as e: logger.error(f"Failed to update workspace {workspace_id}: {str(e)}") raise + + +@router.get("/{workspace_id}/jobs", response_model=list[Job]) +async def get_jobs_by_workspace_id( + workspace_id: int, + repository: JobRepository = Depends(get_jobs_repository), + current_user: UserInfo = Depends(validate_token), +) -> list[Job]: + try: + jobs = await repository.getWorkspaceJobs(current_user, workspace_id) + return jobs + except Exception as e: + logger.error(f"Failed to fetch jobs for workspace {workspace_id}: {str(e)}") + raise diff --git a/docker-compose.local.yml b/docker-compose.local.yml index 8bfeae3..fe4bd88 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -8,7 +8,7 @@ services: ports: - 5432:5432 volumes: - - ./data/db:/var/lib/postgresql/data + - /Users/nareshd/Documents/wa-proviso/docker-data/db:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres -d workspaces-osm-local"] interval: 10s From b841004fe402af58b32c747fef6ee003fd6c5119 Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Wed, 12 Aug 2026 16:21:18 +0530 Subject: [PATCH 2/9] added import job creation on workspace creation Added import job creation on workspace creation --- api/src/workspaces/jobs/repository.py | 61 +++++++++++++++++++-------- api/src/workspaces/routes.py | 40 ++++++++++++++++-- 2 files changed, 80 insertions(+), 21 deletions(-) diff --git a/api/src/workspaces/jobs/repository.py b/api/src/workspaces/jobs/repository.py index 6388f95..8440664 100644 --- a/api/src/workspaces/jobs/repository.py +++ b/api/src/workspaces/jobs/repository.py @@ -24,10 +24,12 @@ def _accessible_workspace_ids(current_user: UserInfo) -> list[int]: return workspace_ids async def create(self, current_user: UserInfo, job_data: JobCreate) -> Job: - if not current_user.isWorkspaceContributor(job_data.workspace_id): - raise ForbiddenException( - "User does not have permissions to create a job in that workspace." - ) + # if not current_user.isWorkspaceContributor(job_data.workspace_id): + # raise ForbiddenException( + # "User does not have permissions to create a job in that workspace." + # ) + # Not needed as this may be during the initial creation. There is no API call to create a job. + # The job is created internally when a workspace is created. So, we don't need to check for permissions here. job = Job(**job_data.model_dump()) @@ -60,25 +62,38 @@ async def update( current_user: UserInfo, job_id: int, job_data: JobPatch, + ignore_permissions: bool = False, ) -> Job: - accessible_workspace_ids = self._accessible_workspace_ids(current_user) - - query = ( - update(Job) - .where( - (Job.id == job_id) - & (Job.workspace_id.in_(accessible_workspace_ids)) # type: ignore[attr-defined] + if ignore_permissions: + query = ( + update(Job) + .where(Job.id == job_id) + .values(**job_data.model_dump(exclude_unset=True)) + ) + result = await self.session.execute(query) + if result.rowcount != 1: # type: ignore[attr-defined] + raise NotFoundException(f"Update failed for job id {job_id}") + await self.session.commit() + return await self._getJobById(job_id) + else: + accessible_workspace_ids = self._accessible_workspace_ids(current_user) + + query = ( + update(Job) + .where( + (Job.id == job_id) + & (Job.workspace_id.in_(accessible_workspace_ids)) # type: ignore[attr-defined] + ) + .values(**job_data.model_dump(exclude_unset=True)) ) - .values(**job_data.model_dump(exclude_unset=True)) - ) - result = await self.session.execute(query) + result = await self.session.execute(query) - if result.rowcount != 1: # type: ignore[attr-defined] - raise NotFoundException(f"Update failed for job id {job_id}") + if result.rowcount != 1: # type: ignore[attr-defined] + raise NotFoundException(f"Update failed for job id {job_id}") - await self.session.commit() - return await self.getById(current_user, job_id) + await self.session.commit() + return await self._getJobById(job_id) async def delete(self, current_user: UserInfo, job_id: int) -> None: accessible_workspace_ids = self._accessible_workspace_ids(current_user) @@ -106,3 +121,13 @@ async def getWorkspaceJobs( query = select(Job).where(Job.workspace_id == workspace_id) result = await self.session.execute(query) return list(result.scalars().all()) + + async def _getJobById(self, job_id: int) -> Job: + query = select(Job).where(Job.id == job_id) + result = await self.session.execute(query) + job = result.scalar_one_or_none() + + if not job: + raise NotFoundException(f"Job with id {job_id} not found") + + return job diff --git a/api/src/workspaces/routes.py b/api/src/workspaces/routes.py index ec25390..3724cb5 100644 --- a/api/src/workspaces/routes.py +++ b/api/src/workspaces/routes.py @@ -17,7 +17,7 @@ from api.src.users.repository import UserRepository from api.src.users.schemas import WorkspaceUserRoleType from api.src.workspaces.jobs.repository import JobRepository -from api.src.workspaces.jobs.schemas import Job +from api.src.workspaces.jobs.schemas import Job, JobCreate, JobPatch from api.src.workspaces.repository import WorkspaceRepository from api.src.workspaces.schemas import ( ImagerySettingsPatch, @@ -201,9 +201,43 @@ async def create_workspace( # workspace and lead role rather than serving stale data for up to # an hour: # - evict_user_from_cache(current_user.user_uuid) + # Get the user access_token from the header + access_token = current_user.credentials + request_data = { + "workspace_id": workspace.id, + "tdei_dataset_id": ( + str(workspace_data.tdeiRecordId) + if workspace_data.tdeiRecordId is not None + else "" + ), + "tdei_token": access_token, + } + create_job = await jobs_repository.create( + current_user, + JobCreate( + job_type="workspace-import", + status="requested", + request=request_data, + workspace_id=workspace.id, + ), + ) + job_id = create_job.id + logger.info( + f"Import job with ID: {job_id} created for workspace ID: {workspace.id}" + ) - return {"workspaceId": workspace.id} + request_data["unique_job_id"] = job_id + # since this is the first time, we donot have the job_id unless we created one. Update the same in the request + await jobs_repository.update( + current_user, + job_id, + JobPatch(request=request_data), + ignore_permissions=True, + ) + # Send the message over the bus here. + + evict_user_from_cache(current_user.user_uuid) + return {"workspaceId": workspace.id, "importJobId": job_id} except Exception as e: logger.error(f"Failed to create workspace: {str(e)}") raise From 3a203bfb179245478d474dbb62740b5228f30996 Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Wed, 12 Aug 2026 17:55:48 +0530 Subject: [PATCH 3/9] added provisioning for proxy to get the data for OSM --- api/core/config.py | 4 +++ api/core/messenger.py | 18 ++++++++++ api/src/workspaces/routes.py | 5 +++ dev/lighttpd.conf | 66 ++++++++++++++++++++++++++++++++++++ docker-compose.local.yml | 10 ++++++ pyproject.toml | 1 + uv.lock | 38 +++++++++++++++++++++ 7 files changed, 142 insertions(+) create mode 100644 api/core/messenger.py create mode 100644 dev/lighttpd.conf diff --git a/api/core/config.py b/api/core/config.py index c6554c7..acbc8e5 100644 --- a/api/core/config.py +++ b/api/core/config.py @@ -73,6 +73,10 @@ class Settings(BaseSettings): SENTRY_DSN: str = "" + # Azure Service Bus connection string and topic name for sending messages + SERVICE_BUS_CONNECTION_STRING: str = "" + SERVICE_BUS_TOPIC_NAME: str = "" + @property def cors_origins_list(self) -> list[str]: """Allowed CORS origins as a list. diff --git a/api/core/messenger.py b/api/core/messenger.py new file mode 100644 index 0000000..1b1d441 --- /dev/null +++ b/api/core/messenger.py @@ -0,0 +1,18 @@ +import json + +from azure.servicebus import ServiceBusClient, ServiceBusMessage + +from api.core.config import settings + + +class Messenger: + def __init__(self): + self.connection_string = settings.SERVICE_BUS_CONNECTION_STRING + self.topic_name = settings.SERVICE_BUS_TOPIC_NAME + + def send_message(self, message: dict): + with ServiceBusClient.from_connection_string(self.connection_string) as client: + sender = client.get_topic_sender(topic_name=self.topic_name) + with sender: + service_bus_message = ServiceBusMessage(json.dumps(message)) + sender.send_messages(service_bus_message) diff --git a/api/src/workspaces/routes.py b/api/src/workspaces/routes.py index 3724cb5..1fd4624 100644 --- a/api/src/workspaces/routes.py +++ b/api/src/workspaces/routes.py @@ -10,6 +10,7 @@ validate_quest_definition_schema, ) from api.core.logging import get_logger +from api.core.messenger import Messenger from api.core.security import UserInfo, evict_user_from_cache, validate_token from api.src.osm.repository import OSMRepository from api.src.osm.routes import get_osm_repo @@ -235,6 +236,10 @@ async def create_workspace( ignore_permissions=True, ) # Send the message over the bus here. + messenger = Messenger() + messenger.send_message( + request_data + ) # send the message to the bus for processing evict_user_from_cache(current_user.user_uuid) return {"workspaceId": workspace.id, "importJobId": job_id} diff --git a/dev/lighttpd.conf b/dev/lighttpd.conf new file mode 100644 index 0000000..cdcff66 --- /dev/null +++ b/dev/lighttpd.conf @@ -0,0 +1,66 @@ +# default document-root +server.document-root = "/app/htdocs" + +# TCP port +server.port = 80 + +# selecting modules +server.modules = ( "mod_access", "mod_rewrite", "mod_fastcgi", "mod_proxy", "mod_alias","mod_openssl" ) + +# handling unknown routes +server.error-handler-404 = "/dispatch.map" + +# read configuration from output of a command +#include_shell "/usr/local/bin/confmimetype /etc/mime.types" +mimetype.assign = ( + ".html" => "text/html", + ".txt" => "text/plain", + ".jpg" => "image/jpeg", + ".png" => "image/png" +) + + +$HTTP["request-method"] == "GET" { + url.rewrite-once = ( + "^/api/0\.6/map(\.(json|xml))?(\?(.*))?$" => "/dispatch.map", + "^/api/0\.6/(node|way|relation)/[[:digit:]]+(\.(json|xml))?$" => "/dispatch.map", + "^/api/0\.6/(node|way|relation)/[[:digit:]]+/history.*$" => "/dispatch.map", + "^/api/0\.6/(node|way|relation)/[[:digit:]]+/[[:digit:]]+.*$" => "/dispatch.map", + "^/api/0\.6/(node|way|relation)/[[:digit:]]+/relations$" => "/dispatch.map", + "^/api/0\.6/node/[[:digit:]]+/ways$" => "/dispatch.map", + "^/api/0\.6/(way|relation)/[[:digit:]]+/full$" => "/dispatch.map", + "^/api/0\.6/changeset/[[:digit:]]+.*$" => "/dispatch.map", + "^/api/0\.6/(nodes|ways|relations)(\?(.*))?$" => "/dispatch.map", + "^/api/0\.6/changeset/[[:digit:]]+/download$" => "/dispatch.map", + ) +} + +$HTTP["request-method"] == "POST" { + url.rewrite-once = ( + "^/api/0\.6/changeset/[[:digit:]]+/upload.*$" => "/dispatch.map", + ) +} + +$HTTP["request-method"] == "PUT" { + url.rewrite-once = ( + "^/api/0\.6/changeset/[[:digit:]]+/close.*$" => "/dispatch.map", + "^/api/0\.6/changeset/[[:digit:]]+$" => "/dispatch.map", + "^/api/0\.6/changeset/create.*$" => "/dispatch.map", + ) +} + +$HTTP["url"] =~ "^/(?!(dispatch\.map))" { + proxy.server = ( "" => ( ( "host" => "osm-rails", "port" => "3000" ) ) ) +} + +fastcgi.debug = 1 + +# For use with Dockerfile +# +fastcgi.server = ( ".map" => + (( "host" => "osm-cgimap", + "port" => 8000, + "check-local" => "disable", + + )) +) diff --git a/docker-compose.local.yml b/docker-compose.local.yml index fe4bd88..f2e7ed0 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -56,6 +56,16 @@ services: tty: true command: ["bundle", "exec", "rake", "jobs:work"] + osm-proxy: + depends_on: + - osm-rails + - osm-cgimap + # 💡 Tip: Change this image to your custom one containing your lighttpd.conf + image: sebp/lighttpd:latest + ports: + - "80:80" + volumes: + - ./dev/lighttpd.conf:/etc/lighttpd/lighttpd.conf:ro backend: build: . container_name: workspaces-backend diff --git a/pyproject.toml b/pyproject.toml index fbb068c..5b05c05 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dependencies = [ "sqlmodel>=0.0.8", "cachetools", "shapely>=2.1.2", + "azure-servicebus>=7.14.3", ] [project.optional-dependencies] diff --git a/uv.lock b/uv.lock index c8e01b6..c85d8ea 100644 --- a/uv.lock +++ b/uv.lock @@ -109,6 +109,33 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/da/21/26f1680ec3a598ea31768f9ebcd427e42986d077a005416094b580635532/autoflake-2.3.3-py3-none-any.whl", hash = "sha256:a51a3412aff16135ee5b3ec25922459fef10c1f23ce6d6c4977188df859e8b53", size = 17715, upload-time = "2026-02-20T05:01:42.137Z" }, ] +[[package]] +name = "azure-core" +version = "1.41.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "requests" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a6/f3/b416179e408990df5db0d516283022dde0f5d0111d98c1a848e41853e81c/azure_core-1.41.0.tar.gz", hash = "sha256:f46ff5dfcd230f25cf1c19e8a34b8dc08a337b2503e268bb600a16c00db8ad5a", size = 381042, upload-time = "2026-05-07T23:30:54.302Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5b/db/325c6d7312d2200251c52323878281045aaffcb5586612296484e4280eaa/azure_core-1.41.0-py3-none-any.whl", hash = "sha256:522b4011e8180b1a3dcd2024396a4e7fe9ac37fb8597db47163d230b5efe892d", size = 220920, upload-time = "2026-05-07T23:30:56.357Z" }, +] + +[[package]] +name = "azure-servicebus" +version = "7.14.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "azure-core" }, + { name = "isodate" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a8/d2/a5c11d4c955e2875de2383c1af8e43ce4899e8418b22e61d32d2bf103626/azure_servicebus-7.14.3.tar.gz", hash = "sha256:70a63384557aec0bee727740e7b25ded29e9e701b77611764577fd7402389402", size = 534786, upload-time = "2025-10-31T05:30:03.342Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/e9/d9fd0b2bef14d85b408c51802142b1c8b7bc3ab08514c89432547b1d87d3/azure_servicebus-7.14.3-py3-none-any.whl", hash = "sha256:386f8d32dae8881661ec8d791c38978eca2bbf7ea9f489d6cff8ad9cc6990234", size = 412522, upload-time = "2025-10-31T05:30:05.252Z" }, +] + [[package]] name = "bcrypt" version = "4.0.1" @@ -680,6 +707,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ef/a6/62565a6e1cf69e10f5727360368e451d4b7f58beeac6173dc9db836a5b46/iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374", size = 5892, upload-time = "2023-01-07T11:08:09.864Z" }, ] +[[package]] +name = "isodate" +version = "0.7.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/54/4d/e940025e2ce31a8ce1202635910747e5a87cc3a6a6bb2d00973375014749/isodate-0.7.2.tar.gz", hash = "sha256:4cd1aa0f43ca76f4a6c6c0292a85f40b35ec2e43e315b59f06e6d32171a953e6", size = 29705, upload-time = "2024-10-08T23:04:11.5Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/15/aa/0aca39a37d3c7eb941ba736ede56d689e7be91cab5d9ca846bde3999eba6/isodate-0.7.2-py3-none-any.whl", hash = "sha256:28009937d8031054830160fce6d409ed342816b543597cece116d966c6d99e15", size = 22320, upload-time = "2024-10-08T23:04:09.501Z" }, +] + [[package]] name = "isort" version = "5.13.2" @@ -1600,6 +1636,7 @@ dependencies = [ { name = "alembic" }, { name = "asyncpg" }, { name = "autoflake" }, + { name = "azure-servicebus" }, { name = "bcrypt" }, { name = "black" }, { name = "cachetools" }, @@ -1639,6 +1676,7 @@ requires-dist = [ { name = "alembic", specifier = ">=1.14.0" }, { name = "asyncpg", specifier = ">=0.30.0" }, { name = "autoflake", specifier = ">=2.3.1" }, + { name = "azure-servicebus", specifier = ">=7.14.3" }, { name = "bcrypt", specifier = "==4.0.1" }, { name = "black", specifier = ">=26.3.1" }, { name = "cachetools" }, From fedde02af45849721113a46992c579498a0e5347 Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Thu, 13 Aug 2026 10:49:26 +0530 Subject: [PATCH 4/9] import status added added import Status to the workspace --- .../aa17ec83af0d_workspace_import_status.py | 31 +++++++++++++++++++ api/src/workspaces/schemas.py | 5 +++ 2 files changed, 36 insertions(+) create mode 100644 alembic_task/versions/aa17ec83af0d_workspace_import_status.py diff --git a/alembic_task/versions/aa17ec83af0d_workspace_import_status.py b/alembic_task/versions/aa17ec83af0d_workspace_import_status.py new file mode 100644 index 0000000..572f059 --- /dev/null +++ b/alembic_task/versions/aa17ec83af0d_workspace_import_status.py @@ -0,0 +1,31 @@ +"""workspace import status + +Revision ID: aa17ec83af0d +Revises: d4e8f1a92b56 +Create Date: 2026-08-13 04:58:59.982773 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "aa17ec83af0d" +down_revision: Union[str, None] = "d4e8f1a92b56" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.add_column("workspaces", sa.Column("importStatus", sa.String(), nullable=True)) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column("workspaces", "importStatus") + # ### end Alembic commands ### diff --git a/api/src/workspaces/schemas.py b/api/src/workspaces/schemas.py index aa2bc0d..96d920e 100644 --- a/api/src/workspaces/schemas.py +++ b/api/src/workspaces/schemas.py @@ -336,3 +336,8 @@ class Workspace(SQLModel, table=True): "cascade": "all, delete-orphan", } ) + + importStatus: Optional[str] = Field( + default=None, + sa_column=Column(Unicode, nullable=True), + ) From 8074243e40383e272431e63eb89b692bcf4f5b3c Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Thu, 13 Aug 2026 16:06:05 +0530 Subject: [PATCH 5/9] Update schemas.py added importStatus to workspaces --- api/src/workspaces/schemas.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/api/src/workspaces/schemas.py b/api/src/workspaces/schemas.py index 96d920e..747efc5 100644 --- a/api/src/workspaces/schemas.py +++ b/api/src/workspaces/schemas.py @@ -227,6 +227,7 @@ class WorkspaceResponse(SQLModel): # this when the app fetches these from dedicated endpoints: longFormQuestDef: Optional[Any] = None imageryListDef: Optional[Any] = None + importStatus: Optional[str] = None # @test: Test that this class properly serializes the workspace data for API responses, including the effective role for the user making the request # @test: Test that the values are populated in this class match the expected values from the database and that the relationships are correctly serialized @@ -264,6 +265,7 @@ def from_workspace( membersCount=members_count, imageryListDef=imagery_list_def, longFormQuestDef=long_form_quest_def, + importStatus=workspace.importStatus, ) From f08cd18e2223ff5ad13497fcb423a0a0b9d61b9c Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Thu, 13 Aug 2026 16:17:20 +0530 Subject: [PATCH 6/9] Update test_workspaces.py Test cases uploaded --- tests/integration/test_workspaces.py | 32 +++++++++++++++++++++++++--- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/tests/integration/test_workspaces.py b/tests/integration/test_workspaces.py index 49a9a79..d654874 100644 --- a/tests/integration/test_workspaces.py +++ b/tests/integration/test_workspaces.py @@ -18,6 +18,7 @@ import api.src.workspaces.routes as ws_routes from api.src.users.schemas import WorkspaceUserRoleType +from api.src.workspaces.jobs.schemas import Job from api.src.workspaces.schemas import QuestDefinitionType, WorkspaceLongQuest from tests.support import factories, fakes @@ -184,9 +185,33 @@ async def test_get_bbox_no_nodes_404(client, login, task_session, osm_session): # === POST "" create ======================================================== -async def test_create_workspace(client, login, task_session, osm_session, evictions): +async def test_create_workspace( + client, login, task_session, osm_session, evictions, monkeypatch +): + class _FakeMessenger: + def send_message(self, _message): + return None + + monkeypatch.setattr(ws_routes, "Messenger", _FakeMessenger) + login(factories.make_user_info(project_group_ids=[factories.DEFAULT_PG_ID])) - osm_session.queue(fakes.scalar(1)) # assign_member_role: user exists + osm_session.queue( + fakes.scalar(1), # assign_member_role: user exists + fakes.affected(1), # assign_member_role: role upsert execute + fakes.affected(1), # jobs_repository.update(...) + fakes.rows( + Job( + id=1, + job_type="workspace-import", + status="requested", + request={"workspace_id": 1}, + current_task=None, + current_task_status=None, + response=None, + workspace_id=1, + ) + ), # jobs_repository._getJobById(...) + ) response = await client.post( f"{API}", @@ -199,8 +224,9 @@ async def test_create_workspace(client, login, task_session, osm_session, evicti assert response.status_code == 201 assert response.json()["workspaceId"] is not None + assert response.json()["importJobId"] is not None assert task_session.commits == 1 # workspace insert - assert osm_session.commits == 1 # role insert + assert osm_session.commits == 3 # role insert + job create + job update assert evictions == [UUID(factories.DEFAULT_USER_ID)] # creator's cache evicted From 5952ef6ca928ac72307df2a964d8caa0266a0be0 Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Thu, 13 Aug 2026 16:27:28 +0530 Subject: [PATCH 7/9] Update repository.py --- api/src/workspaces/repository.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/api/src/workspaces/repository.py b/api/src/workspaces/repository.py index 1dc1941..ba9bceb 100644 --- a/api/src/workspaces/repository.py +++ b/api/src/workspaces/repository.py @@ -18,6 +18,7 @@ WorkspaceImagery, WorkspaceLongQuest, WorkspacePatch, + WorkspaceType, ) @@ -29,10 +30,17 @@ def __init__(self, session: AsyncSession): async def create( self, current_user: UserInfo, workspace_data: WorkspaceCreate ) -> Workspace: + importStatus = "NA" + if ( + workspace_data.type == WorkspaceType.OSW + and workspace_data.tdeiRecordId is not None + ): + importStatus = "in-progress" workspace = Workspace( **workspace_data.model_dump(), createdBy=current_user.user_uuid, # type: ignore[reportArgumentType] createdByName=current_user.user_name, + importStatus=importStatus, ) if str(workspace.tdeiProjectGroupId) not in current_user.getProjectGroupIds(): From 2e3bb2ffdaba49042046e0cb9103990957f239a3 Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Thu, 13 Aug 2026 16:37:20 +0530 Subject: [PATCH 8/9] Fixed the test cases Fixed the test cases --- api/src/workspaces/repository.py | 5 +- api/src/workspaces/routes.py | 76 ++++++++++++++-------------- api/src/workspaces/schemas.py | 6 +++ tests/integration/test_workspaces.py | 41 ++++++++++----- 4 files changed, 74 insertions(+), 54 deletions(-) diff --git a/api/src/workspaces/repository.py b/api/src/workspaces/repository.py index ba9bceb..39a125e 100644 --- a/api/src/workspaces/repository.py +++ b/api/src/workspaces/repository.py @@ -31,10 +31,7 @@ async def create( self, current_user: UserInfo, workspace_data: WorkspaceCreate ) -> Workspace: importStatus = "NA" - if ( - workspace_data.type == WorkspaceType.OSW - and workspace_data.tdeiRecordId is not None - ): + if workspace_data.isTDEIDataset(): importStatus = "in-progress" workspace = Workspace( **workspace_data.model_dump(), diff --git a/api/src/workspaces/routes.py b/api/src/workspaces/routes.py index 1fd4624..b81e57f 100644 --- a/api/src/workspaces/routes.py +++ b/api/src/workspaces/routes.py @@ -202,44 +202,46 @@ async def create_workspace( # workspace and lead role rather than serving stale data for up to # an hour: # - # Get the user access_token from the header - access_token = current_user.credentials - request_data = { - "workspace_id": workspace.id, - "tdei_dataset_id": ( - str(workspace_data.tdeiRecordId) - if workspace_data.tdeiRecordId is not None - else "" - ), - "tdei_token": access_token, - } - create_job = await jobs_repository.create( - current_user, - JobCreate( - job_type="workspace-import", - status="requested", - request=request_data, - workspace_id=workspace.id, - ), - ) - job_id = create_job.id - logger.info( - f"Import job with ID: {job_id} created for workspace ID: {workspace.id}" - ) + job_id = None + if workspace_data.isTDEIOSWDataset(): + # Get the user access_token from the header + access_token = current_user.credentials + request_data = { + "workspace_id": workspace.id, + "tdei_dataset_id": ( + str(workspace_data.tdeiRecordId) + if workspace_data.tdeiRecordId is not None + else "" + ), + "tdei_token": access_token, + } + create_job = await jobs_repository.create( + current_user, + JobCreate( + job_type="workspace-import", + status="requested", + request=request_data, + workspace_id=workspace.id, + ), + ) + job_id = create_job.id + logger.info( + f"Import job with ID: {job_id} created for workspace ID: {workspace.id}" + ) - request_data["unique_job_id"] = job_id - # since this is the first time, we donot have the job_id unless we created one. Update the same in the request - await jobs_repository.update( - current_user, - job_id, - JobPatch(request=request_data), - ignore_permissions=True, - ) - # Send the message over the bus here. - messenger = Messenger() - messenger.send_message( - request_data - ) # send the message to the bus for processing + request_data["unique_job_id"] = job_id + # since this is the first time, we donot have the job_id unless we created one. Update the same in the request + await jobs_repository.update( + current_user, + job_id, + JobPatch(request=request_data), + ignore_permissions=True, + ) + # Send the message over the bus here. + messenger = Messenger() + messenger.send_message( + request_data + ) # send the message to the bus for processing evict_user_from_cache(current_user.user_uuid) return {"workspaceId": workspace.id, "importJobId": job_id} diff --git a/api/src/workspaces/schemas.py b/api/src/workspaces/schemas.py index 747efc5..adc90be 100644 --- a/api/src/workspaces/schemas.py +++ b/api/src/workspaces/schemas.py @@ -139,6 +139,12 @@ class WorkspaceCreate(SQLModel): tdeiServiceId: Optional[UUID] = None tdeiMetadata: Optional[Any] = None + def isTDEIDataset(self) -> bool: + return self.tdeiRecordId is not None and self.tdeiProjectGroupId is not None + + def isTDEIOSWDataset(self) -> bool: + return self.type == WorkspaceType.OSW and self.isTDEIDataset() + class WorkspacePatch(SQLModel): """Fields the client may supply when updating a workspace""" diff --git a/tests/integration/test_workspaces.py b/tests/integration/test_workspaces.py index d654874..2254158 100644 --- a/tests/integration/test_workspaces.py +++ b/tests/integration/test_workspaces.py @@ -186,31 +186,44 @@ async def test_get_bbox_no_nodes_404(client, login, task_session, osm_session): async def test_create_workspace( - client, login, task_session, osm_session, evictions, monkeypatch + client, app, login, task_session, osm_session, evictions, monkeypatch ): class _FakeMessenger: def send_message(self, _message): return None - monkeypatch.setattr(ws_routes, "Messenger", _FakeMessenger) - - login(factories.make_user_info(project_group_ids=[factories.DEFAULT_PG_ID])) - osm_session.queue( - fakes.scalar(1), # assign_member_role: user exists - fakes.affected(1), # assign_member_role: role upsert execute - fakes.affected(1), # jobs_repository.update(...) - fakes.rows( - Job( + class _FakeJobsRepository: + async def create(self, _current_user, job_data): + return Job( id=1, + job_type=job_data.job_type, + status=job_data.status, + request=job_data.request, + current_task=job_data.current_task, + current_task_status=job_data.current_task_status, + response=job_data.response, + workspace_id=job_data.workspace_id, + ) + + async def update(self, _current_user, job_id, job_data, **_kwargs): + return Job( + id=job_id, job_type="workspace-import", status="requested", - request={"workspace_id": 1}, + request=job_data.request if job_data.request is not None else {}, current_task=None, current_task_status=None, response=None, workspace_id=1, ) - ), # jobs_repository._getJobById(...) + + monkeypatch.setattr(ws_routes, "Messenger", _FakeMessenger) + app.dependency_overrides[ws_routes.get_jobs_repository] = _FakeJobsRepository + + login(factories.make_user_info(project_group_ids=[factories.DEFAULT_PG_ID])) + osm_session.queue( + fakes.scalar(1), # assign_member_role: user exists + fakes.affected(1), # assign_member_role: role upsert execute ) response = await client.post( @@ -219,6 +232,8 @@ def send_message(self, _message): "type": "osw", "title": "Fresh", "tdeiProjectGroupId": factories.DEFAULT_PG_ID, + "tdeiRecordId": "33333333-3333-3333-3333-333333333333", + "tdeiServiceId": "44444444-4444-4444-4444-444444444444", }, ) @@ -226,7 +241,7 @@ def send_message(self, _message): assert response.json()["workspaceId"] is not None assert response.json()["importJobId"] is not None assert task_session.commits == 1 # workspace insert - assert osm_session.commits == 3 # role insert + job create + job update + assert osm_session.commits == 1 # role insert assert evictions == [UUID(factories.DEFAULT_USER_ID)] # creator's cache evicted From 15ede529f5c83c466d94424c05e7fd83f043131d Mon Sep 17 00:00:00 2001 From: Naresh Kumar D Date: Thu, 13 Aug 2026 16:53:36 +0530 Subject: [PATCH 9/9] updated tests Updated for tests --- api/src/workspaces/jobs/repository.py | 10 +++++++--- api/src/workspaces/jobs/schemas.py | 14 +++++++++----- api/src/workspaces/routes.py | 2 +- tests/integration/test_workspaces.py | 4 ++++ 4 files changed, 21 insertions(+), 9 deletions(-) diff --git a/api/src/workspaces/jobs/repository.py b/api/src/workspaces/jobs/repository.py index 8440664..fde2640 100644 --- a/api/src/workspaces/jobs/repository.py +++ b/api/src/workspaces/jobs/repository.py @@ -67,7 +67,7 @@ async def update( if ignore_permissions: query = ( update(Job) - .where(Job.id == job_id) + .where(Job.id == job_id) # pyright: ignore[reportArgumentType] .values(**job_data.model_dump(exclude_unset=True)) ) result = await self.session.execute(query) @@ -118,12 +118,16 @@ async def getWorkspaceJobs( "User does not have permissions to view jobs in that workspace." ) - query = select(Job).where(Job.workspace_id == workspace_id) + query = select(Job).where( + Job.workspace_id == workspace_id # pyright: ignore[reportArgumentType] + ) result = await self.session.execute(query) return list(result.scalars().all()) async def _getJobById(self, job_id: int) -> Job: - query = select(Job).where(Job.id == job_id) + query = select(Job).where( + Job.id == job_id # pyright: ignore[reportArgumentType] + ) result = await self.session.execute(query) job = result.scalar_one_or_none() diff --git a/api/src/workspaces/jobs/schemas.py b/api/src/workspaces/jobs/schemas.py index 273e38b..236be45 100644 --- a/api/src/workspaces/jobs/schemas.py +++ b/api/src/workspaces/jobs/schemas.py @@ -7,17 +7,21 @@ class Job(SQLModel, table=True): - __tablename__ = "jobs" + __tablename__ = "jobs" # type: ignore[assignment] id: int = Field(default=None, primary_key=True) job_type: str = Field(default=None, nullable=False) status: str = Field(default=None, nullable=False) - request: dict = Field(default=None, sa_column=Column(SAJson, nullable=False)) + request: dict[str, Any] = Field( + default=None, sa_column=Column(SAJson, nullable=False) + ) created_at: datetime = Field(sa_column=Column(nullable=False, default=datetime.now)) updated_at: datetime = Field(sa_column=Column(nullable=False, default=datetime.now)) - current_task: str = Field(default=None, nullable=True) - current_task_status: str = Field(default=None, nullable=True) - response: dict = Field(default=None, sa_column=Column(SAJson, nullable=True)) + current_task: str | None = Field(default=None, nullable=True) + current_task_status: str | None = Field(default=None, nullable=True) + response: dict[str, Any] | None = Field( + default=None, sa_column=Column(SAJson, nullable=True) + ) workspace_id: int = Field(default=None) # Not sure if we have to add index here. # workspace_id: int = Field(foreign_key="workspaces.id") diff --git a/api/src/workspaces/routes.py b/api/src/workspaces/routes.py index b81e57f..37856d1 100644 --- a/api/src/workspaces/routes.py +++ b/api/src/workspaces/routes.py @@ -183,7 +183,7 @@ async def create_workspace( repository_users: UserRepository = Depends(get_user_repository), jobs_repository: JobRepository = Depends(get_jobs_repository), current_user: UserInfo = Depends(validate_token), -) -> dict[str, int]: +) -> dict[str, int | None]: try: workspace = await repository_ws.create(current_user, workspace_data) assert workspace.id is not None # freshly persisted workspace has an id diff --git a/tests/integration/test_workspaces.py b/tests/integration/test_workspaces.py index 2254158..e5be4b9 100644 --- a/tests/integration/test_workspaces.py +++ b/tests/integration/test_workspaces.py @@ -199,6 +199,8 @@ async def create(self, _current_user, job_data): job_type=job_data.job_type, status=job_data.status, request=job_data.request, + created_at=datetime(2026, 1, 1), + updated_at=datetime(2026, 1, 1), current_task=job_data.current_task, current_task_status=job_data.current_task_status, response=job_data.response, @@ -211,6 +213,8 @@ async def update(self, _current_user, job_id, job_data, **_kwargs): job_type="workspace-import", status="requested", request=job_data.request if job_data.request is not None else {}, + created_at=datetime(2026, 1, 1), + updated_at=datetime(2026, 1, 1), current_task=None, current_task_status=None, response=None,