diff --git a/.bumpversion.toml b/.bumpversion.toml new file mode 100644 index 0000000..24917bb --- /dev/null +++ b/.bumpversion.toml @@ -0,0 +1,15 @@ +[bumpversion] +current_version = "2.1.0rc1+fix-error-handling" +commit = true +tag = true +tag_name = "v{new_version}" + +[[bumpversion.files]] +filename = "pyproject.toml" +search = 'version = "{current_version}"' +replace = 'version = "{new_version}"' + +[[bumpversion.files]] +filename = ".env" +search = 'IMAGE_TAG={current_version}' +replace = 'IMAGE_TAG={new_version}' diff --git a/.env.example b/.env.example index 5d14467..75aa41e 100644 --- a/.env.example +++ b/.env.example @@ -1,10 +1,11 @@ #---- App settings ---- # The API_SERVER_URL is only used to return the complete URL in the result of the job details as specified in OGC. # Should be the base url to the api. - +UMP_SERVER_TIMEOUT=30 UMP_LOG_LEVEL=DEBUG UMP_PROVIDERS_FILE=providers.yaml UMP_API_SERVER_URL=localhost:5000 +UMP_API_SERVER_URL_PREFIX=/api UMP_REMOTE_JOB_STATUS_REQUEST_INTERVAL=5 UMP_DATABASE_NAME=ump UMP_DATABASE_HOST=localhost diff --git a/.github/workflows/deploy-docs.yml b/.github/workflows/deploy-docs.yml deleted file mode 100644 index bd63966..0000000 --- a/.github/workflows/deploy-docs.yml +++ /dev/null @@ -1,60 +0,0 @@ -name: deploy-book - -# Run this when the master or main branch changes -on: - push: - branches: - - master - - main - # If your git repository has the Jupyter Book within some-subfolder next to - # unrelated files, you can make this run only if a file within that specific - # folder has been modified. - # - # paths: - # - some-subfolder/** - -# This job installs dependencies, builds the book, and pushes it to `gh-pages` -jobs: - deploy-book: - runs-on: ubuntu-latest - permissions: - pages: write - id-token: write - steps: - - uses: actions/checkout@v4 - - # Install dependencies - - name: Set up Python 3.11 - uses: actions/setup-python@v5 - with: - python-version: '3.11' - cache: pip # Implicitly uses requirements.txt for cache key - - - name: Install dependencies - run: pip install -r requirements.txt - - # (optional) Cache your executed notebooks between runs - # if you have config: - # execute: - # execute_notebooks: cache - - name: cache executed notebooks - uses: actions/cache@v4 - with: - path: _build/.jupyter_cache - key: jupyter-book-cache-${{ hashFiles('requirements.txt') }} - - # Build the book - - name: Build the book - run: | - jupyter-book build . - - # Upload the book's HTML as an artifact - - name: Upload artifact - uses: actions/upload-pages-artifact@v3 - with: - path: "_build/html" - - # Deploy the book's HTML to GitHub Pages - - name: Deploy to GitHub Pages - id: deployment - uses: actions/deploy-pages@v4 diff --git a/.gitignore b/.gitignore index 134e617..051860b 100644 --- a/.gitignore +++ b/.gitignore @@ -172,3 +172,4 @@ cython_debug/ geoserver_data postgresql_data +scratch/* \ No newline at end of file diff --git a/.vscode/launch.json b/.vscode/launch.json index a11437c..0db9f82 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -5,7 +5,31 @@ "version": "0.2.0", "configurations": [ { - "name": "Launch api", + "name": "ump + db", + "type": "debugpy", + "request": "launch", + "module": "flask", + "envFile": "${workspaceFolder}/.env", + "env": { + "FLASK_APP": "src/ump/main.py", + "FLASK_DEBUG": "1" + }, + "args": [ + "--debug", + "run", + "--no-debugger", + "-p", + // "5005", + "${command:pickArgs}" + ], + "jinja": true, + "autoStartBrowser": false, + "preLaunchTask": "start-postgis-container", + "postDebugTask": "stop-postgis-container", + "justMyCode": false, + }, + { + "name": "ump + db + migrations", "type": "debugpy", "request": "launch", "module": "flask", diff --git a/.vscode/settings.json b/.vscode/settings.json index b4cb618..0143002 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -6,7 +6,7 @@ "--preview" ], "[python]": { - "editor.defaultFormatter": "ms-python.black-formatter", + "editor.defaultFormatter": "charliermarsh.ruff", "editor.formatOnSave": true, "editor.codeActionsOnSave": { "source.organizeImports": "explicit" @@ -20,4 +20,5 @@ ], "ruff.importStrategy": "fromEnvironment", "black-formatter.importStrategy": "fromEnvironment", + "python.analysis.typeCheckingMode": "basic", } \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index d9c845e..440ee01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,43 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +# [2.x] +## [2.1.0] - 2025-07-31 +### Changed: +- improved error handling when requesting remote servers processes and jobs +- helm chart is up-to-date with current UMP +- provider loader listens on any event to be compatible with configmap-updates in k8s +- provider loader improved: debouncing rapid file changes and atomic updates of providers object +- improved server responses in certain situations, especially when something went wrong showing users json information (as this is a json api) in accordance with OGC api process spec +- improved job starting mechanism + +### Added: +- a new setting to control gunicorn worker timout was introduced: UMP_SERVER_TIMEOUT +- a new setting to control ump server path prefix introduced: UMP_API_SERVER_URL_PREFIX +- timeouts for all requests to remote servers + +### Fixed: +- using setting UMP_KEYCLOAK_CLIENT_ID instead of hard-coded "ump-client" +- job insert queries failed when logged-in user created a job +- missing job metadata +- fetch correct job status from remote server + +## [2.0.0] - 2025-06-25 + +### Added +- comprehensive documentation added +- unified database connection pool handling + +### Changed +- created a providers pydantic class for better type safety and concise handling +- improved provider.yaml loading and provider updateing mechanism +- improved logging +- keycloak coinnection error handling improved + +### Fixed +- ump ran out of database connections due to unclosed connections + +# [1.x] ## [1.2.0] - 2024-05-25 ### Added diff --git a/Dockerfile b/Dockerfile index 7631706..db91fdc 100644 --- a/Dockerfile +++ b/Dockerfile @@ -42,8 +42,12 @@ ARG USER_UID=1000 ARG USERNAME=pythonuser ARG USER_GID=2000 ARG SOURCE_COMMIT +ARG IMAGE_TAG=2.0.0 -LABEL maintainer="Urban Data Analytics" name="analytics/urban-model-platform" source_commit=$SOURCE_COMMIT +LABEL maintainer="Urban Data Analytics" \ + name="analytics/urban-model-platform" \ + source_commit=$SOURCE_COMMIT \ + version=${IMAGE_TAG} # add user and group RUN groupadd --gid $USER_GID $USERNAME && \ diff --git a/Makefile b/Makefile index 1c73dd2..ef5a253 100644 --- a/Makefile +++ b/Makefile @@ -49,7 +49,10 @@ initiate-dev: build-image: @echo 'Building release ${CONTAINER_REGISTRY}/${CONTAINER_NAMESPACE}/$(IMAGE_NAME):$(IMAGE_TAG)' # build your image - docker compose -f docker-compose-build.yaml build --build-arg SOURCE_COMMIT=$(GIT_COMMIT) api + docker compose -f docker-compose-build.yaml build \ + --build-arg SOURCE_COMMIT=$(GIT_COMMIT) \ + --build-arg TAG=$(IMAGE_TAG) \ + api upload-image: build-image docker compose -f docker-compose-build.yaml push api @@ -95,3 +98,27 @@ build-docs: clean-docs: jupyter-book clean docs + +# Update app version: bump major, minor, or patch +bump-app-version: + @if [ -z "$(part)" ]; then \ + echo "Usage: make bump-app part={major|minor|patch}"; \ + exit 1; \ + fi; \ + bump-my-version bump $(part) + +# Update app version: set to a specific version +set-app-version: + @if [ -z "$(version)" ]; then \ + echo "Usage: make set-app-version version={version}"; \ + exit 1; \ + fi; \ + bump-my-version set $(version) + +# Update chart version: bump major, minor, or patch +bump-chart-version: + @if [ -z "$(part)" ]; then \ + echo "Usage: make bump-chart part={major|minor|patch}"; \ + exit 1; \ + fi; \ + (cd charts && bump-my-version bump $(part)) \ No newline at end of file diff --git a/charts/.bumpversion.toml b/charts/.bumpversion.toml new file mode 100644 index 0000000..71ba623 --- /dev/null +++ b/charts/.bumpversion.toml @@ -0,0 +1,10 @@ +[bumpversion] +current_version = "0.7.0" +commit = true +tag = true +tag_name = "chart-v{new_version}" + +[[bumpversion.files]] +filename = "urban-model-platform/Chart.yaml" +search = 'version: {current_version}' +replace = 'version: {new_version}' diff --git a/charts/urban-model-platform/Chart.yaml b/charts/urban-model-platform/Chart.yaml index 526f5fb..f38c991 100644 --- a/charts/urban-model-platform/Chart.yaml +++ b/charts/urban-model-platform/Chart.yaml @@ -15,13 +15,13 @@ type: application # This is the chart version. This version number should be incremented each time you make changes # to the chart and its templates, including the app version. # Versions are expected to follow Semantic Versioning (https://semver.org/) -version: 0.6.0 +version: 0.9.5 # This is the version number of the application being deployed. This version number should be # incremented each time you make changes to the application. Versions are not expected to # follow Semantic Versioning. They should reflect the version the application is using. # It is recommended to use it with quotes. -appVersion: "0.9.2" +appVersion: "2.1.0" kubeVersion: ">= 1.31.0" annotations: category: API diff --git a/charts/urban-model-platform/templates/configmap-job-settings.yaml b/charts/urban-model-platform/templates/configmap-job-settings.yaml deleted file mode 100644 index cb34e2b..0000000 --- a/charts/urban-model-platform/templates/configmap-job-settings.yaml +++ /dev/null @@ -1,17 +0,0 @@ -apiVersion: v1 -kind: ConfigMap -metadata: - name: {{ include "ump.fullname" . }}-job-settings - annotations: - "helm.sh/hook": pre-install,pre-upgrade - "helm.sh/hook-weight": "-6" # Lower weight than job (-5) ensures it runs first - "helm.sh/hook-delete-policy": hook-succeeded - labels: - {{- include "ump.labels" . | nindent 4 }} -data: - PROVIDERS_FILE: {{ .Values.config.providersFilePath | quote }} - CORS_URL_REGEX: {{ .Values.config.corsUrlRegex | quote }} - API_SERVER_URL: {{ .Values.config.serverUrl | quote }} - NUMBER_OF_WORKERS: {{ .Values.config.numberOfWorkers | quote }} - POSTGRES_PORT: {{ .Values.config.postgresPort | quote }} - POSTGRES_HOST: {{ .Values.config.postgresHost | quote }} diff --git a/charts/urban-model-platform/templates/configmap-settings.yaml b/charts/urban-model-platform/templates/configmap-settings.yaml index 7b7efeb..b505a90 100644 --- a/charts/urban-model-platform/templates/configmap-settings.yaml +++ b/charts/urban-model-platform/templates/configmap-settings.yaml @@ -5,9 +5,11 @@ metadata: labels: {{- include "ump.labels" . | nindent 4 }} data: - PROVIDERS_FILE: {{ .Values.config.providersFilePath | quote }} - CORS_URL_REGEX: {{ .Values.config.corsUrlRegex | quote }} - API_SERVER_URL: {{ .Values.config.serverUrl | quote }} - NUMBER_OF_WORKERS: {{ .Values.config.numberOfWorkers | quote }} - POSTGRES_PORT: {{ .Values.config.postgresPort | quote }} - POSTGRES_HOST: {{ .Values.config.postgresHost | quote }} + UMP_LOG_LEVEL: {{ .Values.config.logLevel | quote }} + UMP_PROVIDERS_FILE: {{ .Values.config.providersFilePath | quote }} + UMP_API_SERVER_URL: {{ .Values.config.apiServerUrl | quote }} + UMP_API_SERVER_URL_PREFIX: {{ .Values.config.apiServerUrlPrefix | quote }} + UMP_REMOTE_JOB_STATUS_REQUEST_INTERVAL: {{ .Values.config.remoteJobStatusRequestInterval | quote }} + UMP_JOB_DELETE_INTERVAL: {{ .Values.config.jobDeleteInterval | quote }} + UMP_API_SERVER_WORKERS: {{ .Values.config.apiServerWorkers | quote }} + UMP_SERVER_TIMEOUT: {{ .Values.config.serverWorkerTimeout | quote }} diff --git a/charts/urban-model-platform/templates/deployment.yaml b/charts/urban-model-platform/templates/deployment.yaml index 4ffe843..a61043f 100644 --- a/charts/urban-model-platform/templates/deployment.yaml +++ b/charts/urban-model-platform/templates/deployment.yaml @@ -49,70 +49,112 @@ spec: mountPath: /tmp - name: providers-volume mountPath: {{ .Values.config.providersFileMountPath | quote }} - readOnly: true envFrom: - - configMapRef: + - configMapRef: name: {{ include "ump.fullname" . }}-settings + # if keycloak/geoserver secrets are not referenced, they will be created {{- if not .Values.keycloakConnection.existingSecret.name }} - secretRef: name: {{ include "ump.fullname" . }}-keycloak-connection {{- end }} + {{- if not .Values.geoserverConnection.existingSecret.name }} + - secretRef: + name: {{ include "ump.fullname" . }}-geoserver-connection + {{- end }} env: - - name: POSTGRES_DB + - name: FLASK_APP + value: ump.main + # postgres connection + {{- if .Values.postgresConnection.existingSecret.name }} + - name: UMP_DATABASE_NAME valueFrom: secretKeyRef: name: {{ .Values.postgresConnection.existingSecret.name }} key: dbname - {{- if not .Values.config.postgresHost }} - - name: POSTGRES_HOST + - name: UMP_DATABASE_HOST valueFrom: secretKeyRef: name: {{ .Values.postgresConnection.existingSecret.name }} key: host - {{- end }} - {{- if not .Values.config.postgresPort }} - - name: POSTGRES_PORT + - name: UMP_DATABASE_PORT valueFrom: secretKeyRef: name: {{ .Values.postgresConnection.existingSecret.name }} key: port - {{- end }} - - name: POSTGRES_USER + - name: UMP_DATABASE_USER valueFrom: secretKeyRef: name: {{ .Values.postgresConnection.existingSecret.name }} key: user - - name: POSTGRES_PASSWORD + - name: UMP_DATABASE_PASSWORD valueFrom: secretKeyRef: name: {{ .Values.postgresConnection.existingSecret.name }} key: password # keycloak connection + {{- end }} {{- if .Values.keycloakConnection.existingSecret.name }} - - name: KEYCLOAK_USER + - name: UMP_KEYCLOAK_USER valueFrom: secretKeyRef: name: {{ .Values.keycloakConnection.existingSecret.name }} key: user - - name: KEYCLOAK_PASSWORD + - name: UMP_KEYCLOAK_PASSWORD valueFrom: secretKeyRef: name: {{ .Values.keycloakConnection.existingSecret.name }} key: password - - name: KEYCLOAK_HOST + - name: UMP_KEYCLOAK_REALM valueFrom: secretKeyRef: name: {{ .Values.keycloakConnection.existingSecret.name }} - key: host - - name: KEYCLOAK_PROTOCOL + key: realm + - name: UMP_KEYCLOAK_URL valueFrom: secretKeyRef: name: {{ .Values.keycloakConnection.existingSecret.name }} - key: protocol + key: url + - name: UMP_KEYCLOAK_CLIENT_ID + valueFrom: + secretKeyRef: + name: {{ .Values.keycloakConnection.existingSecret.name }} + key: clientId {{- end }} + {{- if .Values.geoserverConnection.existingSecret.name }} + - name: UMP_GEOSERVER_URL + valueFrom: + secretKeyRef: + name: {{ .Values.geoserverConnection.existingSecret.name }} + key: url + - name: UMP_GEOSERVER_DB_HOST + valueFrom: + secretKeyRef: + name: {{ .Values.geoserverConnection.existingSecret.name }} + key: dbHost + - name: UMP_GEOSERVER_DB_PORT + valueFrom: + secretKeyRef: + name: {{ .Values.geoserverConnection.existingSecret.name }} + key: dbPort + - name: UMP_GEOSERVER_WORKSPACE_NAME + valueFrom: + secretKeyRef: + name: {{ .Values.geoserverConnection.existingSecret.name }} + key: workspaceName + - name: UMP_GEOSERVER_USER + valueFrom: + secretKeyRef: + name: {{ .Values.geoserverConnection.existingSecret.name }} + key: user + - name: UMP_GEOSERVER_PASSWORD + valueFrom: + secretKeyRef: + name: {{ .Values.geoserverConnection.existingSecret.name }} + key: password + {{- end }} readinessProbe: httpGet: - path: /api/health/ready + path: {{ .Values.config.apiServerUrlPrefix }}/health/ready port: http initialDelaySeconds: 5 periodSeconds: 10 diff --git a/charts/urban-model-platform/templates/httproute-tls.yaml b/charts/urban-model-platform/templates/httproute-tls.yaml index 27a983e..02a5ee1 100644 --- a/charts/urban-model-platform/templates/httproute-tls.yaml +++ b/charts/urban-model-platform/templates/httproute-tls.yaml @@ -23,4 +23,6 @@ spec: - name: {{ include "ump.fullname" . }} port: {{ .Values.service.port }} kind: Service + group: '' + weight: 1 {{- end -}} \ No newline at end of file diff --git a/charts/urban-model-platform/templates/job-migrations.yaml b/charts/urban-model-platform/templates/job-migrations.yaml deleted file mode 100644 index e873e3a..0000000 --- a/charts/urban-model-platform/templates/job-migrations.yaml +++ /dev/null @@ -1,77 +0,0 @@ -apiVersion: batch/v1 -kind: Job -metadata: - name: {{ include "ump.fullname" . }}-db-migrate - labels: - {{- include "ump.labels" . | nindent 4 }} - annotations: - "helm.sh/hook": pre-install,pre-upgrade - "helm.sh/hook-weight": "-5" - "helm.sh/hook-delete-policy": hook-succeeded -spec: - backoffLimit: 3 - template: - spec: - {{- with .Values.image.pullSecrets }} - imagePullSecrets: - {{- toYaml . | nindent 8 }} - {{- end }} - containers: - - name: db-migrate - image: {{ .Values.image.repository }}:{{ .Values.image.tag }} - command: ["flask", "db", "upgrade"] - volumeMounts: - - name: tmp-volume - mountPath: /tmp - - name: providers-volume - mountPath: {{ .Values.config.providersFilePath | quote }} - subPath: providers.yaml - readOnly: true - envFrom: - - configMapRef: - name: {{ include "ump.fullname" . }}-job-settings - - secretRef: - name: {{ include "ump.fullname" . }}-job-keycloak-connection - env: - - name: POSTGRES_DB - valueFrom: - secretKeyRef: - name: {{ .Values.postgresConnection.existingSecret.name }} - key: dbname - {{- if not .Values.config.postgresHost }} - - name: POSTGRES_HOST - valueFrom: - secretKeyRef: - name: {{ .Values.postgresConnection.existingSecret.name }} - key: host - {{- end }} - {{- if not .Values.config.postgresPort }} - - name: POSTGRES_PORT - valueFrom: - secretKeyRef: - name: {{ .Values.postgresConnection.existingSecret.name }} - key: port - {{- end }} - - name: POSTGRES_USER - valueFrom: - secretKeyRef: - name: {{ .Values.postgresConnection.existingSecret.name }} - key: user - - name: POSTGRES_PASSWORD - valueFrom: - secretKeyRef: - name: {{ .Values.postgresConnection.existingSecret.name }} - key: password - - name: FLASK_APP - value: "ump.main" - restartPolicy: Never - volumes: - - name: tmp-volume - emptyDir: {} - - name: providers-volume - configMap: - {{- if not .Values.providers.existingConfigMap.name }} - name: {{ include "ump.fullname" . }}-providers - {{- else }} - name: {{ .Values.providers.existingConfigMap.name }} - {{- end }} diff --git a/charts/urban-model-platform/templates/role-job-reader.yaml b/charts/urban-model-platform/templates/role-job-reader.yaml deleted file mode 100644 index e6039a1..0000000 --- a/charts/urban-model-platform/templates/role-job-reader.yaml +++ /dev/null @@ -1,10 +0,0 @@ -apiVersion: rbac.authorization.k8s.io/v1 -kind: Role -metadata: - name: {{ include "ump.fullname" . }}-job-reader - labels: - {{- include "ump.labels" . | nindent 4 }} -rules: -- apiGroups: ["batch"] - resources: ["jobs"] - verbs: ["get", "watch", "list"] diff --git a/charts/urban-model-platform/templates/rolebinding-job.yaml b/charts/urban-model-platform/templates/rolebinding-job.yaml deleted file mode 100644 index dd549bc..0000000 --- a/charts/urban-model-platform/templates/rolebinding-job.yaml +++ /dev/null @@ -1,15 +0,0 @@ -# rolebinding-job.yaml -apiVersion: rbac.authorization.k8s.io/v1 -kind: RoleBinding -metadata: - name: {{ include "ump.fullname" . }}-job-reader - labels: - {{- include "ump.labels" . | nindent 4 }} -roleRef: - apiGroup: rbac.authorization.k8s.io - kind: Role - name: {{ include "ump.fullname" . }}-job-reader -subjects: -- kind: ServiceAccount - name: {{ include "ump.serviceAccountName" . }} - namespace: {{ .Release.Namespace }} \ No newline at end of file diff --git a/charts/urban-model-platform/templates/secret-geoserver.yaml b/charts/urban-model-platform/templates/secret-geoserver.yaml new file mode 100644 index 0000000..3c4e746 --- /dev/null +++ b/charts/urban-model-platform/templates/secret-geoserver.yaml @@ -0,0 +1,19 @@ +{{- if not .Values.keycloakConnection.existingSecret.name -}} +apiVersion: v1 +kind: Secret +metadata: + name: {{ include "ump.fullname" . }}-geoserver-connection + labels: + {{ include "ump.labels" . | nindent 4 }} +data: + UMP_GEOSERVER_URL: "aHR0cDovL2dlb3NlcnZlcjo4MDgw" + UMP_GEOSERVER_DB_HOST: "" + UMP_GEOSERVER_DB_PORT: "NTQzMg==" + UMP_GEOSERVER_DB_NAME: "" + UMP_GEOSERVER_DB_USER: "" + UMP_GEOSERVER_DB_PASSWORD: "" + UMP_GEOSERVER_WORKSPACE_NAME: "" + UMP_GEOSERVER_USER: "" + UMP_GEOSERVER_PASSWORD: "" + UMP_GEOSERVER_CONNECTION_TIMEOUT: "MTA=" +{{- end -}} \ No newline at end of file diff --git a/charts/urban-model-platform/templates/secret-keycloak-job.yaml b/charts/urban-model-platform/templates/secret-keycloak-job.yaml deleted file mode 100644 index f470c72..0000000 --- a/charts/urban-model-platform/templates/secret-keycloak-job.yaml +++ /dev/null @@ -1,17 +0,0 @@ -{{- if not .Values.keycloakConnection.existingSecret.name -}} -apiVersion: v1 -kind: Secret -metadata: - name: {{ include "ump.fullname" . }}-job-keycloak-connection - annotations: - "helm.sh/hook": pre-install,pre-upgrade - "helm.sh/hook-weight": "-6" # Lower weight than job (-5) ensures it runs first - "helm.sh/hook-delete-policy": hook-succeeded - labels: - {{ include "ump.labels" . | nindent 4 }} -data: - KEYCLOAK_USER: "" - KEYCLOAK_PASSWORD: "" - KEYCLOAK_HOST: "" - KEYCLOAK_PROTOCOL: "" -{{- end -}} \ No newline at end of file diff --git a/charts/urban-model-platform/templates/secret-keycloak.yaml b/charts/urban-model-platform/templates/secret-keycloak.yaml index bd2a1a8..fc5e264 100644 --- a/charts/urban-model-platform/templates/secret-keycloak.yaml +++ b/charts/urban-model-platform/templates/secret-keycloak.yaml @@ -6,8 +6,9 @@ metadata: labels: {{ include "ump.labels" . | nindent 4 }} data: - KEYCLOAK_USER: "" - KEYCLOAK_PASSWORD: "" - KEYCLOAK_HOST: "" - KEYCLOAK_PROTOCOL: "" + UMP_KEYCLOAK_CLIENT_ID: "" + UMP_KEYCLOAK_PASSWORD: "" + UMP_KEYCLOAK_REALM: "" + UMP_KEYCLOAK_URL: "aHR0cDovL2tleWNsb2FrOjgwODAvYXV0aA==" + UMP_KEYCLOAK_USER: "" {{- end -}} \ No newline at end of file diff --git a/charts/urban-model-platform/values.yaml b/charts/urban-model-platform/values.yaml index 184c096..5f8c2b2 100644 --- a/charts/urban-model-platform/values.yaml +++ b/charts/urban-model-platform/values.yaml @@ -5,7 +5,7 @@ fullnameOverride: "" image: repository: lgvanalytics.azurecr.io/urban-model-platform pullPolicy: IfNotPresent - tag: "dev_improvements" + tag: "2.0.0" pullSecrets: - name: secret @@ -37,18 +37,21 @@ resources: service: type: ClusterIP port: 5000 # port under which the svc answers - targetPort: 5000 port the container itsel uses + targetPort: 5000 # port the container itself uses config: - corsUrlRegex: "*" - serverUrl: "0.0.0.0:5000" + logLevel: "DEBUG" + providersFilePath: "./providers.yaml" providersFileMountPath: /app - providersFilePath: /app/providers.yaml - numberOfWorkers: "1" - postgresHost: "" - postgresPort: 5432 - -postgresConnection: + apiServerUrl: "http://localhost:5000" + apiServerUrlPrefix: "/api" + apiServerWorkers: "4" + remoteJobStatusRequestInterval: "5" + serverWorkerTimeout: "30" # Timeout for server workers, should be higher than all request to remote servers, to avoid server 500 errors + geoserverConnectionTimeout: "60" + jobDeleteInterval: "240" + +postgresConnection: existingSecret: name: postgres-credentials @@ -56,6 +59,10 @@ keycloakConnection: existingSecret: name: "" +geoserverConnection: + existingSecret: + name: "" + # If configMap for providers is already existing and should not be overwritten, set this to true. Default: false providers: existingConfigMap: @@ -86,7 +93,7 @@ autoscaling: targetCPUUtilizationPercentage: 80 targetMemoryUtilizationPercentage: 80 -tolerations: +tolerations: - key: ump/reservedFor operator: "Equal" value: app diff --git a/docker-compose-dev.yaml b/docker-compose-dev.yaml index c9ce2f4..9251c1f 100644 --- a/docker-compose-dev.yaml +++ b/docker-compose-dev.yaml @@ -21,6 +21,7 @@ services: environment: UMP_API_SERVER_URL: localhost:${WEBAPP_PORT_EXTERNAL} UMP_API_SERVER_WORKERS: 1 + UMP_SERVER_TIMEOUT: 30 # seconds UMP_LOG_LEVEL: DEBUG UMP_PROVIDERS_FILE: /home/pythonuser/providers.yaml UMP_KEYCLOAK_USER: admin diff --git a/docs/content/02-user_guide/setup.md b/docs/content/02-user_guide/setup.md index 79c5378..c1b4d73 100644 --- a/docs/content/02-user_guide/setup.md +++ b/docs/content/02-user_guide/setup.md @@ -29,6 +29,7 @@ This document describes the configuration options for the Urban Model Platform ( | `UMP_KEYCLOAK_CLIENT_ID` | Keycloak client ID. | `ump-client` | | `UMP_KEYCLOAK_USER` | Keycloak admin username. | `admin` | | `UMP_KEYCLOAK_PASSWORD` | Keycloak admin password. | `admin` | +| `UMP_API_SERVER_URL_PREFIX` | subpath prefix, e.g.: "/api" | `/` | ### Example Modelserver Settings | Variable | Description | Default Value | diff --git a/poetry.lock b/poetry.lock index db47d24..b08311d 100644 --- a/poetry.lock +++ b/poetry.lock @@ -424,6 +424,39 @@ files = [ {file = "blinker-1.9.0.tar.gz", hash = "sha256:b4ce2265a7abece45e7cc896e98dbebe6cead56bcf805a3d23136d145f5445bf"}, ] +[[package]] +name = "bracex" +version = "2.6" +description = "Bash style brace expander." +optional = false +python-versions = ">=3.9" +files = [ + {file = "bracex-2.6-py3-none-any.whl", hash = "sha256:0b0049264e7340b3ec782b5cb99beb325f36c3782a32e36e876452fd49a09952"}, + {file = "bracex-2.6.tar.gz", hash = "sha256:98f1347cd77e22ee8d967a30ad4e310b233f7754dbf31ff3fceb76145ba47dc7"}, +] + +[[package]] +name = "bump-my-version" +version = "1.2.0" +description = "Version bump your Python project" +optional = false +python-versions = ">=3.8" +files = [ + {file = "bump_my_version-1.2.0-py3-none-any.whl", hash = "sha256:201e6b103ff0f2b240c9d0a6eb83db382840b1f78eb78f6d77726bed39a326d8"}, + {file = "bump_my_version-1.2.0.tar.gz", hash = "sha256:5120d798aaf26468a37ca0f127992dc036688b8e5e106adc8870b13c2a2df22d"}, +] + +[package.dependencies] +click = "*" +httpx = ">=0.28.1" +pydantic = ">=2.0.0" +pydantic-settings = "*" +questionary = "*" +rich = "*" +rich-click = "*" +tomlkit = "*" +wcmatch = ">=8.5.1" + [[package]] name = "certifi" version = "2025.1.31" @@ -4044,6 +4077,20 @@ files = [ [package.dependencies] cffi = {version = "*", markers = "implementation_name == \"pypy\""} +[[package]] +name = "questionary" +version = "2.1.0" +description = "Python library to build pretty command line user prompts ⭐️" +optional = false +python-versions = ">=3.8" +files = [ + {file = "questionary-2.1.0-py3-none-any.whl", hash = "sha256:44174d237b68bc828e4878c763a9ad6790ee61990e0ae72927694ead57bab8ec"}, + {file = "questionary-2.1.0.tar.gz", hash = "sha256:6302cdd645b19667d8f6e6634774e9538bfcd1aad9be287e743d96cacaf95587"}, +] + +[package.dependencies] +prompt_toolkit = ">=2.0,<4.0" + [[package]] name = "referencing" version = "0.36.2" @@ -4113,6 +4160,26 @@ pygments = ">=2.13.0,<3.0.0" [package.extras] jupyter = ["ipywidgets (>=7.5.1,<9)"] +[[package]] +name = "rich-click" +version = "1.8.9" +description = "Format click help output nicely with rich" +optional = false +python-versions = ">=3.7" +files = [ + {file = "rich_click-1.8.9-py3-none-any.whl", hash = "sha256:c3fa81ed8a671a10de65a9e20abf642cfdac6fdb882db1ef465ee33919fbcfe2"}, + {file = "rich_click-1.8.9.tar.gz", hash = "sha256:fd98c0ab9ddc1cf9c0b7463f68daf28b4d0033a74214ceb02f761b3ff2af3136"}, +] + +[package.dependencies] +click = ">=7" +rich = ">=10.7" +typing_extensions = ">=4" + +[package.extras] +dev = ["mypy", "packaging", "pre-commit", "pytest", "pytest-cov", "rich-codex", "ruff", "types-setuptools"] +docs = ["markdown_include", "mkdocs", "mkdocs-glightbox", "mkdocs-material-extensions", "mkdocs-material[imaging] (>=9.5.18,<9.6.0)", "mkdocs-rss-plugin", "mkdocstrings[python]", "rich-codex"] + [[package]] name = "rpds-py" version = "0.23.1" @@ -4256,6 +4323,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:f66efbc1caa63c088dead1c4170d148eabc9b80d95fb75b6c92ac0aad2437d76"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:22353049ba4181685023b25b5b51a574bce33e7f51c759371a7422dcae5402a6"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:932205970b9f9991b34f55136be327501903f7c66830e9760a8ffb15b07f05cd"}, + {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:a52d48f4e7bf9005e8f0a89209bf9a73f7190ddf0489eee5eb51377385f59f2a"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-win32.whl", hash = "sha256:3eac5a91891ceb88138c113f9db04f3cebdae277f5d44eaa3651a4f573e6a5da"}, {file = "ruamel.yaml.clib-0.2.12-cp310-cp310-win_amd64.whl", hash = "sha256:ab007f2f5a87bd08ab1499bdf96f3d5c6ad4dcfa364884cb4549aa0154b13a28"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-macosx_13_0_arm64.whl", hash = "sha256:4a6679521a58256a90b0d89e03992c15144c5f3858f40d7c18886023d7943db6"}, @@ -4264,6 +4332,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:811ea1594b8a0fb466172c384267a4e5e367298af6b228931f273b111f17ef52"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:cf12567a7b565cbf65d438dec6cfbe2917d3c1bdddfce84a9930b7d35ea59642"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:7dd5adc8b930b12c8fc5b99e2d535a09889941aa0d0bd06f4749e9a9397c71d2"}, + {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1492a6051dab8d912fc2adeef0e8c72216b24d57bd896ea607cb90bb0c4981d3"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-win32.whl", hash = "sha256:bd0a08f0bab19093c54e18a14a10b4322e1eacc5217056f3c063bd2f59853ce4"}, {file = "ruamel.yaml.clib-0.2.12-cp311-cp311-win_amd64.whl", hash = "sha256:a274fb2cb086c7a3dea4322ec27f4cb5cc4b6298adb583ab0e211a4682f241eb"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:20b0f8dc160ba83b6dcc0e256846e1a02d044e13f7ea74a3d1d56ede4e48c632"}, @@ -4272,6 +4341,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:749c16fcc4a2b09f28843cda5a193e0283e47454b63ec4b81eaa2242f50e4ccd"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:bf165fef1f223beae7333275156ab2022cffe255dcc51c27f066b4370da81e31"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:32621c177bbf782ca5a18ba4d7af0f1082a3f6e517ac2a18b3974d4edf349680"}, + {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:b82a7c94a498853aa0b272fd5bc67f29008da798d4f93a2f9f289feb8426a58d"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-win32.whl", hash = "sha256:e8c4ebfcfd57177b572e2040777b8abc537cdef58a2120e830124946aa9b42c5"}, {file = "ruamel.yaml.clib-0.2.12-cp312-cp312-win_amd64.whl", hash = "sha256:0467c5965282c62203273b838ae77c0d29d7638c8a4e3a1c8bdd3602c10904e4"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:4c8c5d82f50bb53986a5e02d1b3092b03622c02c2eb78e29bec33fd9593bae1a"}, @@ -4280,6 +4350,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:96777d473c05ee3e5e3c3e999f5d23c6f4ec5b0c38c098b3a5229085f74236c6"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_1_i686.whl", hash = "sha256:3bc2a80e6420ca8b7d3590791e2dfc709c88ab9152c00eeb511c9875ce5778bf"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:e188d2699864c11c36cdfdada94d781fd5d6b0071cd9c427bceb08ad3d7c70e1"}, + {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:4f6f3eac23941b32afccc23081e1f50612bdbe4e982012ef4f5797986828cd01"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-win32.whl", hash = "sha256:6442cb36270b3afb1b4951f060eccca1ce49f3d087ca1ca4563a6eb479cb3de6"}, {file = "ruamel.yaml.clib-0.2.12-cp313-cp313-win_amd64.whl", hash = "sha256:e5b8daf27af0b90da7bb903a876477a9e6d7270be6146906b276605997c7e9a3"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-macosx_12_0_arm64.whl", hash = "sha256:fc4b630cd3fa2cf7fce38afa91d7cfe844a9f75d7f0f36393fa98815e911d987"}, @@ -4288,6 +4359,7 @@ files = [ {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e2f1c3765db32be59d18ab3953f43ab62a761327aafc1594a2a1fbe038b8b8a7"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:d85252669dc32f98ebcd5d36768f5d4faeaeaa2d655ac0473be490ecdae3c285"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:e143ada795c341b56de9418c58d028989093ee611aa27ffb9b7f609c00d813ed"}, + {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:2c59aa6170b990d8d2719323e628aaf36f3bfbc1c26279c0eeeb24d05d2d11c7"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-win32.whl", hash = "sha256:beffaed67936fbbeffd10966a4eb53c402fafd3d6833770516bf7314bc6ffa12"}, {file = "ruamel.yaml.clib-0.2.12-cp39-cp39-win_amd64.whl", hash = "sha256:040ae85536960525ea62868b642bdb0c2cc6021c9f9d507810c0c604e66f5a7b"}, {file = "ruamel.yaml.clib-0.2.12.tar.gz", hash = "sha256:6c8fbb13ec503f99a91901ab46e0b07ae7941cd527393187039aec586fdfd36f"}, @@ -5375,6 +5447,20 @@ files = [ [package.extras] watchmedo = ["PyYAML (>=3.10)"] +[[package]] +name = "wcmatch" +version = "10.1" +description = "Wildcard/glob file name matcher." +optional = false +python-versions = ">=3.9" +files = [ + {file = "wcmatch-10.1-py3-none-any.whl", hash = "sha256:5848ace7dbb0476e5e55ab63c6bbd529745089343427caa5537f230cc01beb8a"}, + {file = "wcmatch-10.1.tar.gz", hash = "sha256:f11f94208c8c8484a16f4f48638a85d771d9513f4ab3f37595978801cb9465af"}, +] + +[package.dependencies] +bracex = ">=2.1.1" + [[package]] name = "wcwidth" version = "0.2.13" @@ -5556,4 +5642,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.0" python-versions = ">=3.11,<3.13" -content-hash = "5b241bf1cb5e085aa28bbe7139e098d802800b08b6ed006f72b9a3e9f3edbe63" +content-hash = "75593a48ecc678f200b1c3203a5cd99e32787d7c65f30258e6cc01545f466093" diff --git a/pyproject.toml b/pyproject.toml index 03eeef3..361d5e1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,7 +38,7 @@ exclude = [ [tool.poetry] name = "ump" -version = "1.2.0" +version = "2.1.0" description = "server federation api, OGC Api Processes-based to connect model servers and centralize access to them" authors = [ "Rico Herzog ", @@ -96,6 +96,7 @@ pre-commit = ">=3.4.0,<4" debugpy = "^1.8.5" flake8 = "^7.1.1" pylint = "^3.3.1" +bump-my-version = "^1.2.0" [tool.poetry.group.docs] optional = true diff --git a/scratch/README.md b/scratch/README.md deleted file mode 100644 index 47fde96..0000000 --- a/scratch/README.md +++ /dev/null @@ -1 +0,0 @@ -Hier kann man kleine Codesnippets ausprobieren und testen. \ No newline at end of file diff --git a/scripts/entrypoint.sh b/scripts/entrypoint.sh index f90290f..26eea13 100755 --- a/scripts/entrypoint.sh +++ b/scripts/entrypoint.sh @@ -1,8 +1,13 @@ #!/bin/bash set -e + +export UMP_SERVER_TIMEOUT="${UMP_SERVER_TIMEOUT:-30}" + +flask db upgrade + echo "Running API Server in production mode." -NUMBER_OF_WORKERS="${NUMBER_OF_WORKERS:-1}" -echo "Running gunicorn with ${NUMBER_OF_WORKERS} workers." +UMP_API_SERVER_WORKERS="${UMP_API_SERVER_WORKERS:-1}" +echo "Running gunicorn with ${UMP_API_SERVER_WORKERS} workers." # export PATH=$PATH:/home/python/.local/bin -exec gunicorn --workers=$NUMBER_OF_WORKERS --bind=0.0.0.0:5000 ump.main:app +exec gunicorn --workers=$UMP_API_SERVER_WORKERS --bind=0.0.0.0:5000 ump.main:app diff --git a/src/ump/api/jobs.py b/src/ump/api/jobs.py index 2bd6a7c..9491248 100644 --- a/src/ump/api/jobs.py +++ b/src/ump/api/jobs.py @@ -3,11 +3,13 @@ from sqlalchemy import select from sqlalchemy.orm import Session +from ump import utils from ump.api.db_handler import DBHandler from ump.api.db_handler import db_engine as engine from ump.api.models.ensemble import Ensemble, JobsEnsembles from ump.api.models.job import Job from ump.api.models.job_status import JobStatus +from ump.config import app_settings def append_ensemble_list(job): with Session(engine) as session: @@ -91,7 +93,11 @@ def next_links(page, limit, count_jobs): links = [] if count_jobs > (page - 1) * limit: links.append({ - "href": f"/api/jobs?page={page+1}&limit={limit}", + "href": utils.join_url_parts( + app_settings.UMP_API_SERVER_URL, + app_settings.UMP_API_SERVER_URL_PREFIX, + f"jobs?page={page+1}&limit={limit}" + ), "rel": "service", "type": "application/json", "hreflang": "en", @@ -100,7 +106,11 @@ def next_links(page, limit, count_jobs): if page > 1: links.append({ - "href": f"/api/jobs?page={page-1}&limit={limit}", + "href": utils.join_url_parts( + app_settings.UMP_API_SERVER_URL, + app_settings.UMP_API_SERVER_URL_PREFIX, + f"jobs?page={page-1}&limit={limit}" + ), "rel": "service", "type": "application/json", "hreflang": "en", diff --git a/src/ump/api/models/job.py b/src/ump/api/models/job.py index d18fafd..20cd756 100644 --- a/src/ump/api/models/job.py +++ b/src/ump/api/models/job.py @@ -7,15 +7,29 @@ import aiohttp import geopandas as gpd +from ump.api.models.ogc_exception import OGCExceptionResponse import ump.api.providers as providers -from ump.config import app_settings as config from ump.api.db_handler import DBHandler from ump.api.models.job_status import JobStatus from ump.api.models.providers_config import ProcessConfig, ProviderConfig -from ump.errors import CustomException, InvalidUsage +from ump.config import app_settings as config +from ump.errors import InvalidUsage, OGCProcessException from ump.geoserver.geoserver import Geoserver +from ump.utils import fetch_json, join_url_parts +results_client_timeout = aiohttp.ClientTimeout( + total=5, # Set a reasonable timeout for the requests + connect=2, # Connection timeout + sock_connect=2, # Socket connection timeout + sock_read=5, # Socket read timeout +) + +# TODO class violates Single Responsibility Principle (SRP), it mixes +# business logic with data access logic and metadata handling +# TODO methods like insert use redundant fields instead of job instance fields +# TODO queries use raw SQL, which is not recommended for different reasons: no migrations, reduced maintainability +# TODO: table schema is missing normalization class Job: DISPLAYED_ATTRIBUTES = [ "processID", @@ -29,7 +43,7 @@ class Job: "updated", "progress", "links", - ] + ] SORTABLE_COLUMNS = [ "created", @@ -62,17 +76,35 @@ def __init__(self, job_id=None, user=None): self.process_id = None self.provider_url = None + # TODO: this produces 404 if a job is beeing queried for which was + # stored with a user id, consider to distinguish between + # 404, 401 and 403 here if job_id and not self._init_from_db(job_id, user): - raise CustomException("Job could not be found!") + raise OGCProcessException( + OGCExceptionResponse( + type="http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/no-such-job", + title="Job not found", + detail="The job with the given ID does not exist.", + status=404, + instance="/".join( + [ + config.UMP_API_SERVER_URL, + f"{config.UMP_API_SERVER_URL_PREFIX}", + "jobs", + job_id + ] + ) + ) + ) - def create( + def insert( self, job_id=None, remote_job_id=None, process_id_with_prefix=None, process_title=None, name=None, - parameters=None, + exec_body=None, user=None, process_version=None, ): @@ -82,20 +114,57 @@ def create( process_id_with_prefix, process_title, name, - parameters, + exec_body, user_id=user, process_version=process_version, ) + # TODO: these metadata should come from remote job + # instead of being set here + # because the remote server ultimately decides if a job was accepted! self.status = JobStatus.accepted.value self.created = datetime.now(timezone.utc) self.updated = datetime.now(timezone.utc) + # TODO: we need proper normalization here + # TODO: we need a SQL model for this, raw queries are error prone query = """ INSERT INTO jobs - (job_id, remote_job_id, process_id, provider_prefix, provider_url, status, progress, parameters, message, created, started, finished, updated, user_id, process_title, name, process_version, hash) + ( + job_id, remote_job_id, process_id, + provider_prefix, provider_url, status, + progress, parameters, message, created, + started, finished, updated, user_id, + process_title, name, process_version, hash + ) VALUES - (%(job_id)s, %(remote_job_id)s, %(process_id)s, %(provider_prefix)s, %(provider_url)s, %(status)s, %(progress)s, %(parameters)s, %(message)s, %(created)s, %(started)s, %(finished)s, %(updated)s, %(user_id)s, %(process_title)s, %(name)s, %(process_version)s, encode(sha512((%(parameters)s :: json :: text || %(process_version)s || %(user_id)s) :: bytea), 'base64')) + ( + %(job_id)s, + %(remote_job_id)s, + %(process_id)s, + %(provider_prefix)s, + %(provider_url)s, + %(status)s, + %(progress)s, + %(parameters)s, + %(message)s, + %(created)s, + %(started)s, + %(finished)s, + %(updated)s, + %(user_id)s, + %(process_title)s, + %(name)s, + %(process_version)s, + encode( + sha512( + convert_to( + %(parameters)s :: json :: text || %(process_version)s || %(user_id)s, + 'UTF8' + ) :: bytea + ), 'base64' + ) + ) """ with DBHandler() as db: db.run_query(query, query_params=self._to_dict()) @@ -134,13 +203,15 @@ def _set_attributes( match = re.search(r"(.*):(.*)", self.process_id_with_prefix) if not match: raise InvalidUsage( - f"Process ID {self.process_id_with_prefix} is not known! " + - "Please check endpoint api/processes for a list of available processes." + f"Process ID {self.process_id_with_prefix} is not known! " + + "Please check endpoint api/processes for a list of available processes." ) self.provider_prefix = match.group(1) self.process_id = match.group(2) - self.provider_url = providers.get_providers()[self.provider_prefix].server_url + self.provider_url = providers.get_providers()[ + self.provider_prefix + ].server_url if not self.job_id: self.job_id = str(uuid.uuid4()) @@ -165,8 +236,8 @@ def _init_from_db(self, job_id, user): def _init_from_dict(self, data): self.job_id = data["job_id"] self.remote_job_id = data["remote_job_id"] - self.process_id = data["process_id"] - self.provider_prefix = data["provider_prefix"] + self.process_id: str = data["process_id"] + self.provider_prefix: str = data["provider_prefix"] self.provider_url = data["provider_url"] self.process_id_with_prefix = f"{data['provider_prefix']}:{data['process_id']}" self.status = data["status"] @@ -205,7 +276,7 @@ def _to_dict(self): "process_version": self.process_version, } - def save(self): + def update(self): self.updated = datetime.now(timezone.utc) query = """ @@ -231,9 +302,11 @@ def set_results_metadata(self, results_as_json): values = [] for column in maximal_values_dict: - data_type = str(types[column]) - if data_type == "float64" and results_df[column].apply(float.is_integer).all(): + if ( + data_type == "float64" + and results_df[column].apply(float.is_integer).all() + ): data_type = "int" values.append( @@ -263,7 +336,7 @@ def set_results_metadata(self, results_as_json): return self.results_metadata - def display(self,additional_metadata=False): + def display(self, additional_metadata=False): job_dict = self._to_dict() job_dict["type"] = "process" job_dict["jobID"] = job_dict.pop("job_id") @@ -272,7 +345,6 @@ def display(self,additional_metadata=False): job_dict["processID"] = self.process_id_with_prefix job_dict["links"] = [] - for attr in job_dict: if isinstance(job_dict[attr], datetime): job_dict[attr] = job_dict[attr].strftime("%Y-%m-%dT%H:%M:%S.%fZ") @@ -282,21 +354,24 @@ def display(self,additional_metadata=False): JobStatus.running.value, JobStatus.accepted.value, ): - - job_result_url = f"{config.UMP_API_SERVER_URL}/api/jobs/{self.job_id}/results" + job_result_url = join_url_parts( + config.UMP_API_SERVER_URL, + config.UMP_API_SERVER_URL_PREFIX, + "jobs", + f"{self.job_id}/results" + ) job_dict["links"] = [ { "href": job_result_url, - "rel": "service", + "rel": "http://www.opengis.net/def/rel/ogc/1.0/results", "type": "application/json", "hreflang": "en", - "title": f"Results of job {self.job_id} as geojson" + - " - available when job is finished.", + "title": "Job result", } ] if isinstance(additional_metadata, str): - additional_metadata = additional_metadata.lower() == "true" + additional_metadata = additional_metadata.lower() == "true" if additional_metadata: metadata = {} @@ -312,57 +387,56 @@ def display(self,additional_metadata=False): metadata["process_version"] = self.process_version if self.process_version is not None: metadata["user_id"] = self.user_id - + job_dict["metadata"] = metadata - for key in ["name", "parameters", "results_metadata", "process_title", "process_version","user_id"]: - job_dict.pop(key, None) + for key in [ + "name", + "parameters", + "results_metadata", + "process_title", + "process_version", + "user_id", + ]: + job_dict.pop(key, None) return job_dict - + else: return {k: job_dict[k] for k in self.DISPLAYED_ATTRIBUTES} - async def results(self): if self.status != JobStatus.successful.value: - return { - "error": f"No results available. Job status = {self.status}.", - "message": self.message, - } + self.results_not_available() provider: ProviderConfig = providers.get_providers()[self.provider_prefix] self.provider_url = provider.server_url - async with aiohttp.ClientSession() as session: + async with aiohttp.ClientSession(timeout=results_client_timeout) as session: auth = providers.authenticate_provider(provider) - response = await session.get( - f"{self.provider_url}jobs/{self.remote_job_id}/results?f=json", - auth=auth, + results = await fetch_json( + session, + url=f"{self.provider_url}jobs/{self.remote_job_id}/results?f=json", + json=self.parameters, headers={ "Content-type": "application/json", "Accept": "application/json", }, + auth=auth ) - if response.status == 200: - return await response.json() - else: - raise CustomException( - "Could not retrieve results from model server " + - f"{self.provider_url} - {response.status}: {response.reason}" - ) + return results async def results_to_geoserver(self): try: provider: ProviderConfig = providers.get_providers()[self.provider_prefix] - process_config: ProcessConfig = provider.processes[self.process_id] + process_config: ProcessConfig = provider.processes[self.process_id] results = await self.results() - if result_path:= process_config.result_path: - parts =result_path.split('.') + if result_path := process_config.result_path: + parts = result_path.split(".") for part in parts: results = results[part] @@ -374,7 +448,9 @@ async def results_to_geoserver(self): logging.info( " --> Successfully stored results for job %s (=%s)/%s to geoserver.", - self.process_id_with_prefix, self.process_id, self.job_id + self.process_id_with_prefix, + self.process_id, + self.job_id, ) except Exception as e: @@ -386,6 +462,53 @@ async def results_to_geoserver(self): e, ) + def results_not_available(self): + """ + Raises an OGCProcessException with a meaningful type and detail + according to the current job status. + """ + status_map = { + JobStatus.failed.value: { + "type": "http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/failed", + "title": "Job failed", + "detail": self.message or "The job failed and no results are available.", + }, + JobStatus.dismissed.value: { + "type": "http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/dismissed", + "title": "Job dismissed", + "detail": self.message or "The job was dismissed and no results are available.", + }, + JobStatus.running.value: { + "type": "http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/result-not-ready", + "title": "Job still running", + "detail": "The job is still running. Results are not yet available.", + }, + JobStatus.accepted.value: { + "type": "http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/result-not-ready", + "title": "Job accepted", + "detail": "The job has been accepted but has not started yet. Results are not available.", + }, + } + + info = status_map.get( + self.status if self.status is not None else "", + { + "type": "http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/no-such-job", + "title": "No results available", + "detail": "No results are available for this job.", + }, + ) + + raise OGCProcessException( + OGCExceptionResponse( + type=info["type"], + title=info["title"], + detail=info["detail"], + status=404, + instance=f"{config.UMP_API_SERVER_URL}/{config.UMP_API_SERVER_URL_PREFIX}/jobs/{self.job_id}/results", + ) + ) + def __str__(self): return f""" ----- src.job.Job ----- diff --git a/src/ump/api/models/ogc_exception.py b/src/ump/api/models/ogc_exception.py new file mode 100644 index 0000000..35b43fb --- /dev/null +++ b/src/ump/api/models/ogc_exception.py @@ -0,0 +1,10 @@ +from pydantic import BaseModel +from typing import Optional, Dict, Any + +class OGCExceptionResponse(BaseModel): + type: str + title: str + status: int + detail: str + instance: Optional[str] = None + additional: Optional[Dict[str, Any]] = None diff --git a/src/ump/api/models/process.py b/src/ump/api/models/process.py index 0aba40f..3a892cc 100644 --- a/src/ump/api/models/process.py +++ b/src/ump/api/models/process.py @@ -2,7 +2,6 @@ import json import logging import re -import time from datetime import datetime, timezone from multiprocessing import dummy @@ -10,98 +9,205 @@ from flask import g import ump.api.providers as providers + +from ump.config import app_settings from ump.api.db_handler import engine from ump.api.models.job import Job, JobStatus +from ump.api.models.ogc_exception import OGCExceptionResponse from ump.api.models.providers_config import ProcessConfig, ProviderConfig from ump.config import app_settings as config -from ump.errors import CustomException, InvalidUsage +from ump.errors import InvalidUsage, OGCProcessException +from ump.utils import fetch_json, fetch_response_content -#TODO: unify logging setup, because this takes not into -# account that an admin wants to configure log level -logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) +client_timeout = aiohttp.ClientTimeout( + total=5, # Set a reasonable timeout for the requests + connect=2, # Connection timeout + sock_connect=2, # Socket connection timeout + sock_read=5, # Socket read timeout +) -class Process: - def __init__(self, process_id_with_prefix=None): +# TODO: this is not an OGC API Process in a strict sense, +# instead it incorporates remote process execution logic +# these two things should probably be separated with regard to +# **Single Responsibility Principle (SRP)** +class Process: + def __init__(self, process_id_with_prefix): self.inputs: dict self.outputs: dict self.process_id_with_prefix = process_id_with_prefix - self.process_title = None + self.id = None + self.title = None self.version = None + self.job_control_options = None + self.description = None + self.keywords = None + self.metadata = None + self.links = None match = re.search(r"([^:]+):(.*)", self.process_id_with_prefix) + if not match: - raise InvalidUsage( - "Process ID %s is not known! Please check endpoint api/processes " - + "for a list of available processes.", + logger.warning( + "Process ID '%s' does not match pattern 'provider:process_id'. " + "See /api/processes for a list of available processes.", self.process_id_with_prefix, ) + raise OGCProcessException( + OGCExceptionResponse( + type="http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/no-such-process", + title="Invalid Process ID", + status=400, + detail=( + f"Process ID '{self.process_id_with_prefix}' " + "does not match pattern: 'provider:process_id'}. " + "See /api/processes for a list of available processes." + ), + instance=f"/processes/{self.process_id_with_prefix}", + ) + ) self.provider_prefix = match.group(1) self.process_id = match.group(2) - if not providers.check_process_availability( - self.provider_prefix, self.process_id - ): - raise InvalidUsage( - "Process ID %s is not known! Please check endpoint api/processes " - + "for a list of available processes.", + # this checks if the process is known from providers and configured as available + # for what purpose? -> security! -if a user selects a remote process which is + # not (!) explicitly configured to be available, but exists on the remote server + try: + process_config = providers.get_process_config( + self.provider_prefix, self.process_id + ) + except ValueError as e: + logger.error(e) + + raise OGCProcessException( + OGCExceptionResponse( + type="http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/no-such-process", + title="Invalid Process ID", + status=400, + detail=( + f"Process ID '{self.process_id_with_prefix}' " + "does not match pattern: 'provider:process_id'}. " + "See /api/processes for a list of available processes." + ), + instance=f"/processes/{self.process_id_with_prefix}", + ) + ) + if process_config.exclude: + logger.warning( + "Process ID '%s' is explicitly excluded. ", self.process_id_with_prefix, ) + # raise OGCProcessException to inform users, but with less detail + raise OGCProcessException( + OGCExceptionResponse( + type="http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/no-such-process", + title="Not available.", + status=400, + detail=( + f"Process ID '{self.process_id_with_prefix}' is not available." + ), + instance=f"/processes/{self.process_id_with_prefix}", + ) + ) + + logger.debug( + "Process %s is available. Loading process details.", + self.process_id_with_prefix, + ) - process_config = providers.get_providers()[self.provider_prefix].processes[self.process_id] - # if anonymous access isn’t enabled, require a token - restricted_access = not getattr(process_config, "anonymous_access", False) + if not process_config.anonymous_access: + logger.debug( + "Process %s requires authentication. Checking user roles.", + self.process_id_with_prefix, + ) - if restricted_access: + # check if token was provided by the user auth = getattr(g, "auth_token", None) - role = f"{self.provider_prefix}_{self.process_id}" - realm_roles = auth.get("realm_access", {}).get("roles", []) if auth else [] - client_roles = (auth.get("resource_access", {})\ - .get("ump-client", {})\ - .get("roles", [])) if auth else [] - allowed = any(r in realm_roles + client_roles for r in [self.provider_prefix, role]) - if not auth or not allowed: - raise InvalidUsage( - "Process ID %s is not known! Please check endpoint api/processes " - + "for a list of available processes.", + + if not auth: + logger.warning( + "User is not allowed to access process %s. " + "The request lacks necessary authentication details!", self.process_id_with_prefix, ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Unauthorized.", + status=401, + detail=( + "The request lacks necessary authentication details." + ), + instance=f"/processes/{self.process_id_with_prefix}", + ) + ) + + role_name = f"{self.provider_prefix}_{self.process_id}" + + realm_roles = auth.get("realm_access", {}).get("roles", []) if auth else [] + + client_roles = ( + ( + auth.get( + "resource_access", {} + ).get( + app_settings.UMP_KEYCLOAK_CLIENT_ID, {} + ).get( + "roles", []) + ) + if auth + else [] + ) - asyncio.run(self.set_details()) - - async def set_details(self): - provider = providers.get_providers()[self.provider_prefix] - # Check for Authentification - auth = providers.authenticate_provider(provider) - - async with aiohttp.ClientSession() as session: - response = await session.get( - f"{provider.server_url}processes/{self.process_id}", - auth=auth, - headers={ - "Content-type": "application/json", - "Accept": "application/json", - }, + allowed = any( + r in realm_roles + client_roles for r in [self.provider_prefix, role_name] ) - if response.status != 200: - # TODO: need to differentiate errors better and give more information(url!) - # widespread usage of InvalidUsage error is impractical, because it - # catches UMP user errors as well as programming errors - raise InvalidUsage( - f"Model/process not found! {response.status}: {response.reason}. " - + "Check /api/processes endpoint for available models/processes.", + if not allowed: + logger.warning( + "User is not allowed to access process %s. " + "User roles: %s, required roles: %s", + self.process_id_with_prefix, + realm_roles + client_roles, + [self.provider_prefix, role_name], + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Forbidden.", + status=403, + detail=( + f"User is not allowed to access process " + f"'{self.process_id_with_prefix}'" + ), + instance=f"/processes/{self.process_id_with_prefix}", + ) ) - process_details = await response.json() + asyncio.run(self.load_process_details()) + + async def load_process_details(self): + provider_config = providers.get_providers()[self.provider_prefix] + + # return auth (BasicAuth curently, only) + # TODO: add support for other auth methods: JWT, ... + auth = providers.authenticate_provider(provider_config) + + async with aiohttp.ClientSession(timeout=client_timeout) as session: + process_details = await fetch_json( + session=session, + url=f"{provider_config.server_url}processes/{self.process_id}", + auth=auth, + ) for key in process_details: setattr(self, key, process_details[key]) - def validate_params(self, parameters): + def validate_exec_body(self, parameters): if not self.inputs: return @@ -120,7 +226,7 @@ def validate_params(self, parameters): payload={"parameter_description": parameter_metadata}, ) else: - logging.warning( + logger.warning( "Model execution %s started without parameter %s.", self.process_id_with_prefix, input, @@ -200,12 +306,20 @@ def check_for_cache(self, parameters, user_id): # p = providers.PROVIDERS[self.provider_prefix]["processes"][self.process_id] provider: ProviderConfig = providers.get_providers()[self.provider_prefix] process_config: ProcessConfig = provider.processes[self.process_id] - + if process_config.deterministic: return None sql = """ - select job_id from jobs where hash = encode(sha512((%(parameters)s :: json :: text || %(process_version)s || %(user_id)s) :: bytea), 'base64') + select job_id from jobs where hash = encode( + sha512( + convert_to( + %(parameters)s :: json :: text || %(process_version)s || %(user_id)s, + 'UTF8' + ), + ), + 'base64' + ) """ with engine.begin() as conn: result = conn.exec_driver_sql( @@ -220,229 +334,285 @@ def check_for_cache(self, parameters, user_id): return row.job_id return None - def execute(self, parameters, user): + def execute(self, exec_body, user): provider: ProviderConfig = providers.get_providers()[self.provider_prefix] - self.validate_params(parameters) + # TODO: make this optional (remote servers should do this themselves) + self.validate_exec_body(exec_body) - logging.info( - " --> Executing %s on model server %s with params %s as process %s for user %s", + logger.info( + "Executing %s on model server %s with params %s as process %s for user %s", self.process_id, str(provider.server_url), - parameters, + exec_body, self.process_id_with_prefix, user, ) - job = asyncio.run(self.start_process_execution(parameters, user)) + job = asyncio.run(self.start_process_execution(exec_body, user)) - _process = dummy.Process(target=self._wait_for_results_async, args=[job]) - _process.start() + results_thread = dummy.Process(target=self._wait_for_results_async, args=[job]) + results_thread.start() - result = {"jobID": job.job_id, "status": job.status} - return result + return {"jobID": job.job_id, "status": job.status} async def start_process_execution(self, request_body, user): - # execution mode: - # to maintain backwards compatibility to models using - # pre-1.0.0 versions of OGC api processes - #TODO: need to add prefer-header request_body["mode"] = "async" provider: ProviderConfig = providers.get_providers()[self.provider_prefix] - - # extract job_name from request_body - name = request_body.pop("job_name",None) + name = request_body.pop("job_name", None) job_id = self.check_for_cache(request_body, user) if job_id: - logging.info("Job found, returning cached job.") + logger.info("Job found, returning cached job.") return Job(job_id, user) - # TODO: this try...except block is too big! - # also it does not catch the exact error from the backend server in some cases - # only telling the user that he has a Bad request - try: - auth = providers.authenticate_provider(provider) - - async with aiohttp.ClientSession() as session: - process_response = await session.get( - f"{provider.server_url}processes/{self.process_id}", - auth=auth, - headers={ - "Content-type": "application/json", - "Accept": "application/json", - }, - ) - response = await session.post( - f"{provider.server_url}processes/{self.process_id}/execution", - json=request_body, - auth=auth, - headers={ - "Content-type": "application/json", - "Accept": "application/json", - # execution mode shall be async, if model supports it - "Prefer": "respond-async", - }, - ) - - process_response.raise_for_status() - - if process_response.ok: - process_details = await process_response.json() - self.process_title = process_details["title"] - - response.raise_for_status() - - if response.ok and response.headers: - # Retrieve the job id from the simulation model server from the - # location header: - match = re.search( - "http.*/jobs/(.*)$", response.headers["location"] - ) or (re.search(".*/jobs/(.*)$", response.headers["location"])) - if match: - remote_job_id = match.group(1) - - job = Job() - job.create( - remote_job_id=remote_job_id, - process_id_with_prefix=self.process_id_with_prefix, - process_title=self.process_title, - name=name, - parameters=request_body, - user=user, - process_version=self.version, - ) - job.started = datetime.now(timezone.utc).strftime( - "%Y-%m-%dT%H:%M:%S.%fZ" - ) - - status_response = await session.get( - f"{provider.server_url}jobs/{remote_job_id}?f=json", - auth=auth, - headers={ - "Content-type": "application/json", - "Accept": "application/json", - }, - ) - status_response.raise_for_status() - status_json = await status_response.json() + auth = providers.authenticate_provider(provider) - job.status = status_json.get("status") - job.save() + async with aiohttp.ClientSession(timeout=client_timeout) as session: + try: + response = await self._submit_remote_job( + session, str(provider.server_url), request_body, auth + ) - logging.info( - " --> Job %s for model %s started running.", - job.job_id, - self.process_id_with_prefix, - ) + response_content = await fetch_response_content(response) - return job + response.raise_for_status() # Raise an error for bad responses - except Exception as e: - raise CustomException(f"Job could not be started remotely: {e}") from e + remote_job_id = await self._extract_remote_job_id(response) - def _wait_for_results_async(self, job: Job): - asyncio.run(self._wait_for_results(job)) + job = await self._create_local_job_instance( + remote_job_id, name, request_body, user + ) - async def _wait_for_results(self, job: Job): + # this can probably be omitted here and deferred for _wait_for_status + remote_job_status_info = await self._fetch_remote_job_status( + session, provider.server_url, remote_job_id, auth + ) - logging.info(" --> Waiting for results in Thread") + self._update_job_from_status( + job, remote_job_status_info + ) - finished = False - - provider: ProviderConfig = providers.get_providers()[self.provider_prefix] - - timeout = float(provider.timeout) - start = time.time() - job_details = {} + job.update() - try: - while not finished: + logger.info( + "Job %s for model %s started.", + job.job_id, + self.process_id_with_prefix, + ) - async with aiohttp.ClientSession() as session: + return job + + except aiohttp.ClientResponseError as e: + logger.error("HTTP error during job submission: %s", e) + # response_body = await response.text() + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Remote job submission failed", + status=e.status, + detail=( + "Job could not be started remotely " + f"due to {response_content[0]}" + ), + instance=f"/processes/{self.process_id_with_prefix}/jobs", + ) + ) from e + + except Exception as e: + logger.exception("Unexpected error during job submission: \n%s", e) + + raise OGCProcessException( + OGCExceptionResponse( + type="http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/server-error", + title="Unexpected error during job submission", + status=500, + detail=f"Job could not be started remotely due to unexpected error. Please see logs for more details.", + instance=f"/processes/{self.process_id_with_prefix}/execution", + ) + ) from e + + async def _submit_remote_job( + self, + session: aiohttp.ClientSession, + url: str, + request_body: dict, + auth: aiohttp.BasicAuth | None, + ) -> aiohttp.ClientResponse: + + response = await session.post( + f"{url}processes/{self.process_id}/execution", + json=request_body, + auth=auth, + headers={ + "Content-type": "application/json", + "Accept": "application/json", + "Prefer": "respond-async", + }, + ) - auth = providers.authenticate_provider(provider) - async with session.get( - f"{provider.server_url}jobs/{job.remote_job_id}?f=json", - auth=auth, - headers={ - "Content-type": "application/json", - "Accept": "application/json", - }, - ) as response: + if response.headers or response.content_type == "application/json": + return response + + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="No valid response from remote server.", + status=502, + detail=( + "No application/json response and no headers from " + "remote server received. Response is not valid." + ), + instance=f"/processes/{self.process_id_with_prefix}/jobs", + ) + ) - response.raise_for_status() - job_details: dict = await response.json() + async def _create_local_job_instance( + self, remote_job_id: str, name, request_body, user + ): + job = Job() + + job.insert( + remote_job_id=remote_job_id, + process_id_with_prefix=self.process_id_with_prefix, + process_title=self.title, + name=name, + exec_body=request_body, + user=user, + process_version=self.version, + ) - finished = self.is_finished(job_details) + return job + + async def _fetch_remote_job_status( + self, session: aiohttp.ClientSession, url, remote_job_id, auth + ) -> dict: + job_status = await fetch_json( + session=session, + url=f"{url}jobs/{remote_job_id}?f=json", + auth=auth, + headers={ + "Content-type": "application/json", + "Accept": "application/json", + }, + ) - logging.info(" --> Current Job status: %s", str(job_details)) + return job_status - # either remote job has progress info or else we cannot provide it either - job.progress = job_details.get("progress") + async def _extract_remote_job_id(self, response: aiohttp.ClientResponse) -> str: + """ + Extracts the remote job ID from the aiohttp response object. + Tries the Location header first, then the response body for jobID/jobId. + """ + location_header = response.headers.get("location") + if location_header: + match = re.search(r"/jobs/([^/]+)", location_header) + if match: + return match.group(1) - job.updated = datetime.now(timezone.utc).strftime( - "%Y-%m-%dT%H:%M:%S.%fZ" - ) - job.save() + # Try to extract jobID from response body + try: + if response.content_type == "application/json": + body_json = await response.json() + job_id = body_json.get("jobID") or body_json.get("jobId") + if job_id: + return job_id + except Exception as e: + logger.warning("Failed to parse jobID from response body: %s", e) + + raise OGCProcessException( + OGCExceptionResponse( + type="http://www.opengis.net/def/exceptions/ogcapi-processes-1/1.0/no-such-process", + title="Invalid Location Header", + status=500, + detail="Could not extract remote job ID from Location header or response body.", + instance="/jobs", + ) + ) - if time.time() - start > timeout: - raise TimeoutError( - f"Job did not finish within {timeout/60} minutes. Giving up." - ) + def _wait_for_results_async(self, job: Job): + asyncio.run(self._wait_for_results(job)) - time.sleep(config.UMP_REMOTE_JOB_STATUS_REQUEST_INTERVAL) + async def _wait_for_results(self, job: Job): + logger.info("Thread started to wait for results.") + provider_config: ProviderConfig = providers.get_providers()[ + self.provider_prefix + ] + timeout_seconds = provider_config.timeout - logging.info( - " --> Remote execution job %s: success = %s. Took approx. %s minutes.", - job.remote_job_id, - finished, - int((time.time() - start) / 60), + try: + await asyncio.wait_for( + self._poll_job_until_finished(job, provider_config), + timeout=timeout_seconds, + ) + except asyncio.TimeoutError: + self._set_job_failed( + job, + ( + "While waiting for remote job to finish the" + f" timeout ({provider_config.timeout} sec.) was reached." + ), ) - except Exception as e: - logging.error( - " --> Could not retrieve results for job %s (=%s)/%s from simulation model server: %s", - self.process_id_with_prefix, - self.process_id, - job.job_id, - e, + logger.error("Error while waiting for job results: %s", e) + self._set_job_failed( + job, + ( + "An unexpected error occurred while waiting for job results." + "See the logs for details" + ), ) - job.status = JobStatus.failed.value - job.message = str(e) - job.updated = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") - job.finished = job.updated - job.progress = 100 - job.save() - raise CustomException( - f"Could not retrieve results from simulation model server. {e}" - ) from e - - # Check if job was successful - try: - if job_details["status"] != JobStatus.successful.value: - job.status = JobStatus.failed.value - job.finished = datetime.now(timezone.utc).strftime( - "%Y-%m-%dT%H:%M:%S.%fZ" - ) - job.updated = job.finished - job.progress = 100 - job.message = ( - f'Remote execution was not successful! {job_details["message"]}' - ) - job.save() - raise CustomException(f"Remote job {job.remote_job_id}: {job.message}") - - except CustomException as e: - logging.error(" --> An error occurred: %s", e) - + else: + await self._store_results_if_needed(job) + + async def _poll_job_until_finished(self, job, provider_config): + while True: + status_info = await self._fetch_remote_job_status( + aiohttp.ClientSession(timeout=client_timeout), + provider_config.server_url, + job.remote_job_id, + providers.authenticate_provider(provider_config), + ) + self._update_job_from_status(job, status_info) + if self.is_finished(status_info): + break + await asyncio.sleep(config.UMP_REMOTE_JOB_STATUS_REQUEST_INTERVAL) + + return status_info + + def _update_job_from_status(self, job: Job, status_info): + job.started = status_info.get("started") + job.created = status_info.get("created") + job.updated = status_info.get("updated") + job.finished = status_info.get("finished") + job.message = status_info.get("message", "") + job.progress = status_info.get("progress") + job.status = status_info.get("status", "") + + # save to database + job.update() + + def _set_job_failed(self, job: Job, message: str): + job.status = JobStatus.failed.value + job.message = message + + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + job.finished = now + job.updated = now + job.progress = 0 + job.update() + + # left here if we want to set a job as successful, manually + # this is not used in the current implementation, but could be useful + def _set_job_successful(self, job: Job): job.status = JobStatus.successful.value - job.finished = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") - job.updated = job.finished + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + job.finished = now + job.updated = now job.progress = 100 - job.save() + job.update() - # Check if results should be stored in the geoserver + async def _store_results_if_needed(self, job: Job): try: if ( providers.check_result_storage(self.provider_prefix, self.process_id) @@ -450,15 +620,7 @@ async def _wait_for_results(self, job: Job): ): await job.results_to_geoserver() except Exception as e: - logging.error( - " --> Could not store results for job %s (=%s)/%s to geoserver: %s", - self.process_id_with_prefix, - self.process_id, - job.job_id, - e, - ) - job.message = str(e) - job.save() + logger.error("Could not store results for job %s: %s", job.job_id, e) def is_finished(self, job_details): finished = False diff --git a/src/ump/api/models/providers_config.py b/src/ump/api/models/providers_config.py index a70e49b..f4bbb21 100644 --- a/src/ump/api/models/providers_config.py +++ b/src/ump/api/models/providers_config.py @@ -1,6 +1,9 @@ from typing import Annotated, Literal, TypeAlias -from pydantic import BaseModel, Field, HttpUrl, SecretStr, TypeAdapter, model_validator +from pydantic import ( + BaseModel, Field, HttpUrl, SecretStr, TypeAdapter, + field_validator, model_validator +) # a type alias to give context to an otherwise generic str ProviderName: TypeAlias = Annotated[str, Field( @@ -120,6 +123,14 @@ class ProviderConfig(BaseModel): ) ) + @field_validator("server_url", mode="before") + def ensure_trailing_slash(cls, value: str) -> HttpUrl: + """Ensure server_url has a trailing slash.""" + + if not str(value).endswith("/"): + value += "/" + return HttpUrl(value) + # a TypeAlias to give context to an otherwise generic dict ModelServers: TypeAlias = Annotated[ dict[str, ProviderConfig], @@ -133,7 +144,7 @@ class ProviderConfig(BaseModel): # a TypeAdapter allows us to use pydantics model_validate method # on arbitrary python types -model_servers_adapter = TypeAdapter(ModelServers) +model_servers_adapter: TypeAdapter[ModelServers] = TypeAdapter(ModelServers) if __name__ == "__main__": diff --git a/src/ump/api/processes.py b/src/ump/api/processes.py index 7554732..62c0799 100644 --- a/src/ump/api/processes.py +++ b/src/ump/api/processes.py @@ -3,32 +3,56 @@ from logging import getLogger import aiohttp -from aiohttp import ClientTimeout +from aiohttp import ClientSession, ClientTimeout from flask import g -from ump.api.models.providers_config import ProcessConfig +from ump.config import app_settings +from ump.api.models.providers_config import ProcessConfig, ProviderConfig from ump.api.providers import ( authenticate_provider, get_providers, ) +from ump.errors import OGCProcessException +from ump.utils import fetch_json logger = getLogger(__name__) #TODO: add validation of loaded processes through pydantic model or existing Process class async def load_processes(): processes = [] + auth = g.get("auth_token", {}) or {} - realm_roles = auth.get("realm_access", {}).get("roles", []) - # TODO: another hard-coded one: "ump-client" - client_roles = ( - auth.get("resource_access", {}).get("ump-client", {}).get("roles", []) + + # TODO manually parsing jwt is not recommended, use a library like PyJWT or better Authlib + realm_roles: list = auth.get("realm_access", {}).get("roles", []) + + client_roles: list = ( + auth.get( + "resource_access", {} + ).get( + app_settings.UMP_KEYCLOAK_CLIENT_ID, {} + ).get( + "roles", [] + ) ) - async with aiohttp.ClientSession(raise_for_status=True) as session: + client_timeout = ClientTimeout( + total=5, # Set a reasonable timeout for the requests + connect=2, # Connection timeout + sock_connect=2, # Socket connection timeout + sock_read=5, # Socket read timeout + ) # remote server needs to answer in time, because we make multiple requests! + + async with aiohttp.ClientSession( + raise_for_status=False, timeout=client_timeout + ) as session: # Create a list of tasks for fetching processes concurrently + #TODO: it would make more sense if, not all processes are fetched, + # but only those that are configured and are accessible by the user tasks = [ fetch_provider_processes( - session, provider_name, provider_config, realm_roles, client_roles + session, provider_name, + provider_config, realm_roles, client_roles ) for provider_name, provider_config in get_providers().items() ] @@ -38,7 +62,7 @@ async def load_processes(): # Process results for result in results: - if isinstance(result, Exception): + if isinstance(result, BaseException): logger.error("Error fetching processes: %s", result) else: processes.extend(result) @@ -47,18 +71,25 @@ async def load_processes(): async def fetch_provider_processes( - session, provider_name, provider_config, - realm_roles, client_roles + session: ClientSession, + provider_name: str, provider_config: ProviderConfig, + realm_roles: list, client_roles: list ): """Fetch processes for a specific provider and filter them.""" provider_processes = [] try: provider_auth = authenticate_provider(provider_config) - results = await fetch_processes_from_provider( - session, provider_config, provider_auth + results = await fetch_json( + session=session, + url=f"{provider_config.server_url}processes", + raise_for_status=True, + headers={"Content-type": "application/json", "Accept": "application/json"}, + auth=provider_auth ) + # TODO: instead of manually checking for a key, we should validate the response + # using a pydantic model or json schema! if "processes" in results: for process in results["processes"]: process_id = process["id"] @@ -71,15 +102,24 @@ async def fetch_provider_processes( continue process_config = provider_config.processes[process_id] - if is_process_visible( + if has_user_access_rights( process_id, provider_name, process_config, realm_roles, client_roles ): process["id"] = f"{provider_name}:{process_id}" provider_processes.append(process) + else: + logger.error( + "The response from the remote service was not valid. " + "URL: %s, Content: %s", + provider_config.server_url, + results + ) - except aiohttp.ClientError as e: + # Note: fetch_json raises OGCProcessException on errors + except OGCProcessException as e: logger.error("HTTP error while accessing provider %s: %s", provider_name, e) + except Exception as e: logger.error("Unexpected error while processing provider %s: %s", provider_name, e) traceback.print_exc() @@ -107,7 +147,7 @@ async def fetch_processes_from_provider(session, provider_config, provider_auth) ) raise -def is_process_visible( +def has_user_access_rights( process_id: str, provider_name: str, process_config: ProcessConfig, @@ -141,7 +181,8 @@ def is_process_visible( # Log the specific condition(s) that grant access if process_config.anonymous_access: logger.info( - "Granting access for process %s: Anonymous access is allowed.", + "Granting access for process %s:%s: Anonymous access is allowed.", + provider_name, process_id ) diff --git a/src/ump/api/providers.py b/src/ump/api/providers.py index c54e0f1..d5441a3 100644 --- a/src/ump/api/providers.py +++ b/src/ump/api/providers.py @@ -1,7 +1,9 @@ import atexit -import logging +import time from logging import getLogger -from threading import Lock +from threading import Lock, Timer +import threading +from typing import Optional import aiohttp import yaml @@ -19,33 +21,97 @@ logger = getLogger(__name__) -PROVIDERS: ModelServers = ModelServers() -PROVIDERS_LOCK = Lock() # Thread-safe lock for updating PROVIDERS +# Thread-safe provider storage +PROVIDERS: ModelServers = {} +PROVIDERS_LOCK = Lock() +RELOAD_TIMER: Optional[Timer] = None +DEBOUNCE_DELAY = 0.5 # 500ms debounce class ProviderLoader(FileSystemEventHandler): - def on_modified(self, event): - logger.info("File modified: %s", event.src_path) - if event.src_path == config.UMP_PROVIDERS_FILE.absolute().as_posix(): - self.load_providers() + def __init__(self): + self.last_reload = 0 + self.reload_lock = threading.Lock() + + # need to listen on any event, not just file changes for k8s configmap updates + def on_any_event(self, event): + # Ignore directory events + if event.is_directory: + return + + logger.info("File event: %s on %s", event.event_type, event.src_path) + + event.src_path = str(event.src_path) + + # Check if the event affects our config file + config_path = config.UMP_PROVIDERS_FILE.absolute().as_posix() + + endswith = event.src_path.endswith(config.UMP_PROVIDERS_FILE.name) + contains = '..data' in event.src_path + + if ( + event.src_path == config_path or + endswith or + contains + ): + self._debounced_reload() + + def _debounced_reload(self): + """Debounce rapid file changes to avoid reload storms""" + global RELOAD_TIMER + + # Cancel existing timer + if RELOAD_TIMER: + RELOAD_TIMER.cancel() + + # Schedule debounced reload + RELOAD_TIMER = Timer(DEBOUNCE_DELAY, self.load_providers) + RELOAD_TIMER.start() def load_providers(self): logger.info("(Re)Loading providers from %s", config.UMP_PROVIDERS_FILE) + + # Create new providers dict (don't modify global state yet) + new_providers = {} + try: with open(config.UMP_PROVIDERS_FILE, encoding="UTF-8") as file: if content := yaml.safe_load(file): + # Validate before applying validated_content = model_servers_adapter.validate_python(content) - with PROVIDERS_LOCK: # Ensure thread-safe update - # Update PROVIDERS in place - PROVIDERS.clear() - PROVIDERS.update(validated_content) - logger.info("Providers (re)loaded successfully") + new_providers.update(validated_content) + + # Atomic update with rollback capability + self._atomic_update(new_providers) + logger.info("Providers (re)loaded successfully") + else: + logger.warning("Providers file is empty, keeping current configuration") + except FileNotFoundError: logger.error("Providers file not found: %s", config.UMP_PROVIDERS_FILE) except yaml.YAMLError as e: logger.error("Failed to parse providers file: %s", e) except ValidationError as e: logger.error("Validation error in providers file: %s", e) + except Exception as e: + logger.error("Unexpected error loading providers: %s", e) + + def _atomic_update(self, new_providers: ModelServers): + """Atomically update providers with rollback capability""" + global PROVIDERS + + with PROVIDERS_LOCK: + # Store old providers for potential rollback + old_providers = PROVIDERS + try: + # Create a new dict with copied Pydantic models + PROVIDERS = { + name: provider.model_copy(deep=True) + for name, provider in new_providers.items() + } + except Exception as e: + PROVIDERS = old_providers + raise # Initialize the ProviderLoader and load providers initially @@ -55,30 +121,51 @@ def load_providers(self): observer = PollingObserver() observer.schedule( provider_loader, - config.UMP_PROVIDERS_FILE.absolute(), # Simplified path handling + config.UMP_PROVIDERS_FILE.parent.as_posix(), # Watch directory, not file recursive=False ) observer.start() +def cleanup(): + """Cleanup function for graceful shutdown""" + global RELOAD_TIMER + + if RELOAD_TIMER: + RELOAD_TIMER.cancel() + + observer.stop() + observer.join(timeout=5) # Give it 5 seconds to stop gracefully + # Graceful shutdown for observer -atexit.register(observer.stop) +atexit.register(cleanup) + def get_providers() -> ModelServers: - return PROVIDERS + """Get a copy of current providers (thread-safe and immutable)""" + with PROVIDERS_LOCK: + return PROVIDERS.copy() + + +def get_provider(provider_name: str) -> Optional[ProviderConfig]: + """Get a specific provider by name (thread-safe)""" + with PROVIDERS_LOCK: + return PROVIDERS.get(provider_name) + def authenticate_provider(provider: ProviderConfig): + """Create authentication object for a provider""" auth = None if provider.authentication: auth = aiohttp.BasicAuth( - provider.authentication.user, provider.authentication.password + provider.authentication.user, + provider.authentication.password.get_secret_value() ) return auth -def check_process_availability(provider: str, process_id: str): - available = False - - with PROVIDERS_LOCK: # Ensure thread-safe access +def check_process_availability(provider: str, process_id: str) -> bool: + """Check if a process is available and not excluded""" + with PROVIDERS_LOCK: if ( provider in PROVIDERS and process_id in PROVIDERS[provider].processes @@ -87,16 +174,54 @@ def check_process_availability(provider: str, process_id: str): available = not process.exclude if process.exclude: - logging.debug("Excluding process %s based on configuration", process_id) - - return available + logger.debug("Excluding process %s based on configuration", process_id) + + return available + + return False -def check_result_storage(provider, process_id): - with PROVIDERS_LOCK: # Ensure thread-safe access +def check_result_storage(provider: str, process_id: str) -> Optional[str]: + """Get the result storage type for a process""" + with PROVIDERS_LOCK: if ( provider in PROVIDERS and process_id in PROVIDERS[provider].processes ): return PROVIDERS[provider].processes[process_id].result_storage return None + + +def get_process_config(provider: str, process_id: str) -> ProcessConfig: + """Get complete process configuration""" + with PROVIDERS_LOCK: + if ( + provider in PROVIDERS + and process_id in PROVIDERS[provider].processes + ): + return PROVIDERS[provider].processes[process_id] + + raise ValueError( + f"Process '{process_id}' not found for provider '{provider}'" + ) + + +def list_providers() -> list[str]: + """Get list of all provider names""" + with PROVIDERS_LOCK: + return list(PROVIDERS.keys()) + + +def list_processes(provider: str) -> list[str]: + """Get list of all process IDs for a provider""" + with PROVIDERS_LOCK: + if provider in PROVIDERS: + return list(PROVIDERS[provider].processes.keys()) + return [] + + +# Health check function +def is_healthy() -> bool: + """Check if the provider loader is healthy""" + with PROVIDERS_LOCK: + return len(PROVIDERS) > 0 and observer.is_alive() \ No newline at end of file diff --git a/src/ump/api/routes/jobs.py b/src/ump/api/routes/jobs.py index a2fa726..4e6c935 100644 --- a/src/ump/api/routes/jobs.py +++ b/src/ump/api/routes/jobs.py @@ -7,13 +7,12 @@ from flask import Response, g, request from sqlalchemy import or_, select from sqlalchemy.orm import Session -from ump import config from ump.api.models.ensemble import JobsUsers from ump.api.models.job import Job from ump.api.models.job_comments import JobComment from ump.api.jobs import append_ensemble_list, get_jobs from ump.api.keycloak_utils import find_user_id_by_email -from ump.config import app_settings as config +from ump.api.db_handler import engine jobs = APIBlueprint("jobs", __name__) diff --git a/src/ump/api/routes/processes.py b/src/ump/api/routes/processes.py index 0ecbce9..240c65f 100644 --- a/src/ump/api/routes/processes.py +++ b/src/ump/api/routes/processes.py @@ -27,6 +27,8 @@ def show(process_id_with_prefix=None): def execute(process_id_with_prefix=None): auth = g.get('auth_token') process = Process(process_id_with_prefix) + + # extract unique user ID ('sub') from auth token if available result = process.execute(request.json, None if auth is None else auth['sub']) return Response(json.dumps(result), status=201, mimetype="application/json") diff --git a/src/ump/config.py b/src/ump/config.py index 65fdbf8..fdddc53 100644 --- a/src/ump/config.py +++ b/src/ump/config.py @@ -1,39 +1,46 @@ import logging +from pathlib import Path from pydantic import FilePath, HttpUrl, SecretStr, computed_field, field_validator from pydantic_settings import BaseSettings from rich import print logger = logging.getLogger(__name__) + + # using pydantic_settings to manage environment variables # and do automatic type casting in a central place class UmpSettings(BaseSettings): UMP_LOG_LEVEL: str = "INFO" - UMP_PROVIDERS_FILE: FilePath = "providers.yaml" - UMP_API_SERVER_URL: str = "localhost:3000" + UMP_PROVIDERS_FILE: FilePath = Path("providers.yaml") + UMP_API_SERVER_URL: str = "http://localhost:3000" UMP_API_SERVER_WORKERS: int = 4 UMP_REMOTE_JOB_STATUS_REQUEST_INTERVAL: int = 5 UMP_DATABASE_NAME: str = "ump" UMP_DATABASE_HOST: str = "postgres" UMP_DATABASE_PORT: int = 5432 UMP_DATABASE_USER: str = "postgres" - UMP_DATABASE_PASSWORD: SecretStr = "postgres" + UMP_DATABASE_PASSWORD: SecretStr = SecretStr("postgres") UMP_GEOSERVER_URL: HttpUrl | None = HttpUrl("http://geoserver:8080/geoserver") UMP_GEOSERVER_DB_HOST: str = "postgis" UMP_GEOSERVER_DB_PORT: int = 5432 UMP_GEOSERVER_DB_NAME: str = "ump" UMP_GEOSERVER_DB_USER: str = "ump" - UMP_GEOSERVER_DB_PASSWORD: SecretStr = "ump" + UMP_GEOSERVER_DB_PASSWORD: SecretStr = SecretStr("ump") UMP_GEOSERVER_WORKSPACE_NAME: str = "UMP" UMP_GEOSERVER_USER: str = "geoserver" - UMP_GEOSERVER_PASSWORD: SecretStr = "geoserver" - UMP_GEOSERVER_CONNECTION_TIMEOUT: int = 60 # seconds - UMP_JOB_DELETE_INTERVAL: int = 240 # minutes - UMP_KEYCLOAK_URL: HttpUrl = "http://keycloak:8080/auth" + UMP_GEOSERVER_PASSWORD: SecretStr = SecretStr("geoserver") + UMP_GEOSERVER_CONNECTION_TIMEOUT: int = 60 # seconds + UMP_JOB_DELETE_INTERVAL: int = 240 # minutes + UMP_KEYCLOAK_URL: HttpUrl = HttpUrl("http://keycloak:8080/auth") UMP_KEYCLOAK_REALM: str = "UrbanModelPlatform" UMP_KEYCLOAK_CLIENT_ID: str = "ump-client" UMP_KEYCLOAK_USER: str = "admin" - UMP_KEYCLOAK_PASSWORD: SecretStr = "admin" + UMP_KEYCLOAK_PASSWORD: SecretStr = SecretStr("admin") + UMP_API_SERVER_URL_PREFIX: str = "/" + + # Gunicorn default timeout is 30 seconds + UMP_SERVER_TIMEOUT: int = 30 @computed_field @property @@ -59,5 +66,6 @@ def ensure_trailing_slash(cls, value: str) -> str: value += "/" return value + app_settings = UmpSettings() app_settings.print_settings() diff --git a/src/ump/errors.py b/src/ump/errors.py index e45db62..aaa7a35 100644 --- a/src/ump/errors.py +++ b/src/ump/errors.py @@ -1,6 +1,8 @@ import logging import traceback +from ump.api.models.ogc_exception import OGCExceptionResponse + class CustomException(Exception): status_code = 400 @@ -28,3 +30,7 @@ class InvalidUsage(CustomException): class GeoserverException(CustomException): pass + +class OGCProcessException(Exception): + def __init__(self, response: OGCExceptionResponse): + self.response = response \ No newline at end of file diff --git a/src/ump/geoserver/geoserver.py b/src/ump/geoserver/geoserver.py index 6e91b36..64fc675 100644 --- a/src/ump/geoserver/geoserver.py +++ b/src/ump/geoserver/geoserver.py @@ -58,7 +58,7 @@ def save_results(self, job_id: str, data: dict): try: self.create_workspace() - logging.info(" --> Workspace should be created now") + logging.info(f"Workspace {self.workspace} created now") self.geojson_to_postgis(data=data, table_name=job_id) @@ -76,9 +76,15 @@ def save_results(self, job_id: str, data: dict): def publish_layer(self, store_name: str, layer_name: str): try: response = requests.post( - f"{config.UMP_GEOSERVER_PATH_WORKSPACE}/{self.workspace}" - + f"/datastores/{store_name}/featuretypes", - auth=(config.UMP_GEOSERVER_USER, config.UMP_GEOSERVER_PASSWORD.get_secret_value()), + ( + f"{config.UMP_GEOSERVER_URL_WORKSPACE}/{self.workspace}" + f"/datastores/{store_name}/featuretypes"), + + auth=( + config.UMP_GEOSERVER_USER, + config.UMP_GEOSERVER_PASSWORD.get_secret_value() + ), + data=f"{layer_name}", headers={"Content-type": "text/xml"}, timeout=config.UMP_GEOSERVER_CONNECTION_TIMEOUT, diff --git a/src/ump/main.py b/src/ump/main.py index 51c51a4..418e4f2 100644 --- a/src/ump/main.py +++ b/src/ump/main.py @@ -1,9 +1,10 @@ -#TODO: this file has become a hodgepodge of very different things, +# TODO: this file has become a hodgepodge of very different things, # it should be split into dedicated files: # - an app factory for migrations # - logging setup # - a geoserver cleanup runner -# - the flask app +# - the flask app +# - the token verification import atexit import json import os @@ -17,7 +18,7 @@ from flask_cors import CORS from flask_migrate import Migrate from flask_sqlalchemy import SQLAlchemy -from keycloak import KeycloakOpenID, KeycloakGetError, KeycloakConnectionError +from keycloak import KeycloakConnectionError, KeycloakGetError, KeycloakOpenID from werkzeug.exceptions import HTTPException from werkzeug.middleware.proxy_fix import ProxyFix @@ -29,7 +30,7 @@ from ump.api.routes.processes import processes from ump.api.routes.users import users from ump.config import app_settings as config -from ump.errors import CustomException +from ump.errors import CustomException, OGCProcessException dictConfig( { @@ -46,10 +47,7 @@ "formatter": "default", } }, - "root": { - "level": config.UMP_LOG_LEVEL, - "handlers": ["wsgi"] - }, + "root": {"level": config.UMP_LOG_LEVEL, "handlers": ["wsgi"]}, "loggers": { "ump.api.providers": { # Configure the logger for providers.py "level": config.UMP_LOG_LEVEL, @@ -61,6 +59,11 @@ "handlers": ["wsgi"], "propagate": False, }, + "ump.api.models.process": { # Configure the logger for processes.py + "level": config.UMP_LOG_LEVEL, + "handlers": ["wsgi"], + "propagate": False, + }, }, } ) @@ -69,10 +72,10 @@ def cleanup(): """Cleans up jobs and Geoserver layers of anonymous users""" sql = "delete from jobs where user_id is null and finished < %(finished)s returning job_id, provider_prefix, process_id" - finished = datetime.now() - timedelta(minutes = config.UMP_JOB_DELETE_INTERVAL) - + finished = datetime.now() - timedelta(minutes=config.UMP_JOB_DELETE_INTERVAL) + with DBHandler() as conn: - result = conn.run_query(sql, query_params={'finished': finished}) + result = conn.run_query(sql, query_params={"finished": finished}) for row in result: # get additional job metadata job_id, provider_prefix, process_id = row @@ -90,20 +93,26 @@ def cleanup(): + f"/layers/{job_id}.xml", auth=( config.UMP_GEOSERVER_USER, - config.UMP_GEOSERVER_PASSWORD.get_secret_value + config.UMP_GEOSERVER_PASSWORD.get_secret_value(), ), timeout=config.UMP_GEOSERVER_CONNECTION_TIMEOUT, ) requests.delete( f"{config.UMP_GEOSERVER_URL_WORKSPACE}/{config.UMP_GEOSERVER_WORKSPACE_NAME}" + f"/datastores/{job_id}/featuretypes/{job_id}.xml", - auth=(config.UMP_GEOSERVER_USER, config.UMP_GEOSERVER_PASSWORD.get_secret_value()), + auth=( + config.UMP_GEOSERVER_USER, + config.UMP_GEOSERVER_PASSWORD.get_secret_value(), + ), timeout=config.UMP_GEOSERVER_CONNECTION_TIMEOUT, ) requests.delete( f"{config.UMP_GEOSERVER_URL_WORKSPACE}/{config.UMP_GEOSERVER_WORKSPACE_NAME}" + f"/datastores/{job_id}.xml", - auth=(config.UMP_GEOSERVER_USER, config.UMP_GEOSERVER_PASSWORD.get_secret_value()), + auth=( + config.UMP_GEOSERVER_USER, + config.UMP_GEOSERVER_PASSWORD.get_secret_value(), + ), timeout=config.UMP_GEOSERVER_CONNECTION_TIMEOUT, ) @@ -129,7 +138,7 @@ def cleanup(): CORS(app) -api = APIBlueprint("api", __name__, url_prefix="/api") +api = APIBlueprint("api", __name__, url_prefix=config.UMP_API_SERVER_URL_PREFIX) api.register_blueprint(processes, url_prefix="/processes") api.register_blueprint(jobs, url_prefix="/jobs") api.register_blueprint(ensembles, url_prefix="/ensembles") @@ -146,6 +155,18 @@ def cleanup(): realm_name=config.UMP_KEYCLOAK_REALM, ) + +@app.errorhandler(OGCProcessException) +def handle_ogc_exception(error: OGCProcessException): + response = jsonify(error.response.model_dump(exclude_unset=True)) + response.status_code = error.response.status + response.content_type = "application/problem+json" + + if response.status_code in (401, 403): + response.headers["WWW-Authenticate"] = 'Bearer' + return response + + @app.before_request def check_jwt(): """Decodes the JWT token and runs pending scheduled jobs""" @@ -153,11 +174,14 @@ def check_jwt(): schedule.run_pending() auth = request.authorization + # TODO: this needs to be improved, it should not fail the app + # and error messages must adopt OGCProcessException if auth is not None: # need exception handling here to avoid app failure! try: # TODO: generally token verification is done offline, # but decode_token connects to keycloak on every request (for retrieving keys) + # use a library like PyJWT or better Authlib decoded = keycloak_openid.decode_token(auth.token) except KeycloakGetError as e: app.logger.error(e) @@ -182,6 +206,7 @@ def check_jwt(): g.auth_token = None pass + @app.after_request def set_headers(response): response.headers["Referrer-Policy"] = "no-referrer" @@ -217,5 +242,6 @@ def shutdown_pool_on_exit(): """Close the connection pool when the application shuts down.""" close_pool() + if __name__ == "__main__": app.run(host="0.0.0.0") diff --git a/src/ump/utils.py b/src/ump/utils.py new file mode 100644 index 0000000..4c9e34f --- /dev/null +++ b/src/ump/utils.py @@ -0,0 +1,144 @@ +import asyncio +import logging + +import aiohttp + +from ump.api.models.ogc_exception import OGCExceptionResponse +from ump.errors import OGCProcessException + + +logger = logging.getLogger(__name__) + +def join_url_parts(*parts): + return '/'.join(str(part).strip('/') for part in parts if part != '') + +async def fetch_response_content( + response: aiohttp.ClientResponse +) -> tuple[str | dict, str, int]: + """ + Reads the content of an aiohttp response, handling both JSON and text, + regardless of HTTP status code. + Returns a tuple: (content, content_type, status_code) + """ + content_type = response.headers.get("Content-Type", "") + status = response.status + + try: + if "application/json" in content_type: + content = await response.json() + else: + content = await response.text() + except Exception: + # If JSON parsing fails, fallback to text + content = await response.text() + content_type = "text/plain" + + return content, content_type, status + +# TODO: retry on timeouts, connection errors, etc. +async def fetch_json( + session: aiohttp.ClientSession, url, + raise_for_status=False, **kwargs +) -> dict: + try: + async with session.get(url, **kwargs) as response: + try: + # is the response JSON? + response_data = await response.json() + + except aiohttp.ContentTypeError: + text = await response.text() + logger.error( + "Invalid JSON response from remote service. URL: %s, Content: %s", + url, text[:500] + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Invalid Response Content", + status=502, + detail=( + "The response from the remote service was not " + f"valid JSON: '{text[:100]}'" + ), + instance=None + ) + ) + + # is the response ok? + if raise_for_status: + response.raise_for_status() + + # if all went good, go! + return response_data + + except asyncio.TimeoutError: + logger.error( + "Timeout when requesting remote service. URL: %s", + url + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Upstream Timeout", + status=504, + detail="The request to the remote service timed out.", + instance=None + ) + ) + except aiohttp.ClientResponseError as e: + if e.status == 401: + logger.warning( + "Authentication failed when requesting remote service. URL: %s, Error: %s", + url, str(e) + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Authentication Failed", + status=401, + detail="Authentication with the remote service failed.", + instance=None + ) + ) + logger.error( + "HTTP error when requesting remote service. URL: %s, Status: %s, Error: %s", + url, e.status, str(e) + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Upstream HTTP Error", + status=e.status, + detail=f"The remote service returned an HTTP error: {response_data}", + instance=None + ) + ) + except aiohttp.ClientError as e: + logger.error( + "Connection error when requesting remote service. URL: %s, Error: %s", + url, str(e) + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Upstream Connection Error", + status=502, + detail="There was a connection error with the remote service.", + instance=None + ) + ) + except Exception as e: + logger.error( + "Unexpected error for remote service. URL: %s, Error: %s", + url, str(e) + ) + raise OGCProcessException( + OGCExceptionResponse( + type="about:blank", + title="Internal Server Error", + status=500, + detail="An unexpected error occurred while processing your request.", + instance=None + ) + ) \ No newline at end of file