Release 2024.7.1

.prod.env

@@ -177,3 +177,5 @@ GLOBAL_STORAGE=10737418240
 
 # Gunicorn server socket
 PORT=5000
+
+GEVENT_WORKER=True

server/application.py

@@ -9,14 +9,17 @@
 # Modules that had direct imports (NOT patched): ['urllib3.util, 'urllib3.util.ssl']"
 # which comes from using requests (its deps) lib in webhooks
 
-import os
-from random import randint
+from mergin.config import Configuration as MainConfig
 
-if not os.getenv("NO_MONKEY_PATCH", False):
+if MainConfig.GEVENT_WORKER:
     import gevent.monkey
+    import psycogreen.gevent
+
+    gevent.monkey.patch_all()
+    psycogreen.gevent.patch_psycopg()
 
-    gevent.monkey.patch_all(subprocess=True)
 
+from random import randint
 from celery.schedules import crontab
 from mergin.app import create_app
 from mergin.auth.tasks import anonymize_removed_users

server/config.py

@@ -25,18 +25,6 @@
 """
 import logging
 
-try:
-    from psycogreen.gevent import patch_psycopg
-except ImportError:
-    import sys
-    import traceback
-
-    exception_info = traceback.format_exc()
-    sys.stderr.write(
-        f"Failed to load required functions from the psycogreen library: { exception_info }\n"
-    )
-    sys.exit(1)
-
 worker_class = "gevent"
 
 workers = 2
@@ -59,12 +47,7 @@
 
 max_requests_jitter = 5000
 
-
-def do_post_fork(server, worker):
-    patch_psycopg()
-
-
-post_fork = do_post_fork
+timeout = 30
 
 
 """

server/mergin/__init__.py

@@ -1,5 +1,3 @@
 # Copyright (C) Lutra Consulting Limited
 #
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
-
-from .app import db, mail, ma, create_app

server/mergin/app.py

@@ -7,11 +7,12 @@
 import os
 import connexion
 import wtforms_json
+import gevent
 from marshmallow import fields
 from sqlalchemy.schema import MetaData
 from flask_sqlalchemy import SQLAlchemy
 from flask_marshmallow import Marshmallow
-from flask import json, jsonify, request, abort, current_app, Flask
+from flask import json, jsonify, request, abort, current_app, Flask, Request, Response
 from flask_login import current_user, LoginManager
 from flask_wtf.csrf import generate_csrf, CSRFProtect
 from flask_migrate import Migrate
@@ -27,6 +28,7 @@
 from typing import List, Dict, Optional
 
 from .sync.utils import get_blacklisted_dirs, get_blacklisted_files
+from .config import Configuration
 
 convention = {
     "ix": "ix_%(column_0_label)s",
@@ -105,9 +107,24 @@ def update_obj(self, obj):
                 field.populate_obj(obj, name)
 
 
-def create_simple_app() -> Flask:
-    from .config import Configuration
+class GeventTimeoutMiddleware:
+    """Middleware to implement gevent.Timeout() for all requests"""
+
+    def __init__(self, app):
+        self.app = app
+
+    def __call__(self, environ, start_response):
+        request = Request(environ)
+        try:
+            with gevent.Timeout(Configuration.GEVENT_REQUEST_TIMEOUT):
+                return self.app(environ, start_response)
+        except gevent.Timeout:
+            logging.error(f"Gevent worker: Request {request.path} timed out")
+            resp = Response("Gateway Timeout", mimetype="text/plain", status=504)
+            return resp(environ, start_response)
+
 
+def create_simple_app() -> Flask:
     app = connexion.FlaskApp(__name__, specification_dir=os.path.join(this_dir))
     flask_app = app.app
 
@@ -117,6 +134,9 @@ def create_simple_app() -> Flask:
     ma.init_app(flask_app)
     Migrate(flask_app, db)
     flask_app.connexion_app = app
+    # in case of gevent worker type use middleware to implement custom request timeout
+    if Configuration.GEVENT_WORKER:
+        flask_app.wsgi_app = GeventTimeoutMiddleware(flask_app.wsgi_app)
 
     @flask_app.cli.command()
     def init_db():
@@ -133,7 +153,6 @@ def init_db():
 def create_app(public_keys: List[str] = None) -> Flask:
     """Factory function to create Flask app instance"""
     from itsdangerous import BadTimeSignature, BadSignature
-    from .config import Configuration
     from .auth import auth_required, decode_token
     from .auth.models import User
 

server/mergin/auth/app.py

@@ -12,7 +12,7 @@
 from .commands import add_commands
 from .config import Configuration
 from .models import User, UserProfile
-from .. import db
+from ..app import db
 
 # signal for other versions to listen to
 user_account_closed = signal("user_account_closed")

server/mergin/auth/commands.py

@@ -6,7 +6,7 @@
 from flask import Flask
 from sqlalchemy import or_, func
 
-from .. import db
+from ..app import db
 from .models import User, UserProfile
 
 

server/mergin/auth/controller.py

@@ -35,8 +35,7 @@
     UserChangePasswordForm,
     ApiLoginForm,
 )
-from .. import db
-from ..app import DEPRECATION_API_MSG
+from ..app import DEPRECATION_API_MSG, db
 from ..utils import format_time_delta
 
 

server/mergin/auth/models.py

@@ -9,7 +9,7 @@
 from flask import current_app, request
 from sqlalchemy import or_, func
 
-from .. import db
+from ..app import db
 from ..sync.utils import get_user_agent, get_ip, get_device_id
 
 

server/mergin/auth/schemas.py

@@ -5,9 +5,8 @@
 from flask import current_app
 from marshmallow import fields
 
-from .. import ma
 from .models import User, UserProfile
-from ..app import DateTimeWithZ
+from ..app import DateTimeWithZ, ma
 
 
 class UserProfileSchema(ma.SQLAlchemyAutoSchema):

server/mergin/auth/tasks.py

@@ -6,7 +6,7 @@
 from sqlalchemy.sql.operators import isnot
 
 from ..celery import celery
-from .. import db
+from .app import db
 from .models import User
 from .config import Configuration
 

server/mergin/celery.py

@@ -9,7 +9,7 @@
 from smtplib import SMTPException, SMTPServerDisconnected
 
 from .config import Configuration
-from . import mail
+from .app import mail
 
 
 # create on flask app independent object

server/mergin/config.py

@@ -3,10 +3,9 @@
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
 
 import os
-from .version import get_version
 from decouple import config, Csv
 
-config_dir = os.path.abspath(os.path.dirname(__file__))
+from .version import get_version
 
 
 class Configuration(object):
@@ -95,10 +94,16 @@ class Configuration(object):
 
     # build hash number
     BUILD_HASH = config("BUILD_HASH", default="")
+
     # Allow changing access to admin panel
     ENABLE_SUPERADMIN_ASSIGNMENT = config(
         "ENABLE_SUPERADMIN_ASSIGNMENT", default=True, cast=bool
     )
     # backend version
     VERSION = config("VERSION", default=get_version())
     SERVER_TYPE = config("SERVER_TYPE", default="ce")
+
+    # whether to run flask app with gevent worker type in gunicorn
+    # using gevent type of worker impose some requirements on code, e.g. to be greenlet safe, custom timeouts
+    GEVENT_WORKER = config("GEVENT_WORKER", default=False, cast=bool)
+    GEVENT_REQUEST_TIMEOUT = config("GEVENT_REQUEST_TIMEOUT", default=30, cast=int)

server/mergin/stats/models.py

@@ -5,7 +5,7 @@
 import uuid
 from sqlalchemy.dialects.postgresql import UUID
 
-from .. import db
+from ..app import db
 
 
 class MerginInfo(db.Model):

server/mergin/sync/commands.py

@@ -10,9 +10,10 @@
 from flask import Flask, current_app
 
 from .files import UploadChanges
-from .. import db
+from ..app import db
 from .models import Project, ProjectAccess, ProjectVersion
 from .utils import split_project_path
+from ..auth.models import User
 
 
 def add_commands(app: Flask):
@@ -24,10 +25,21 @@ def project():
     @project.command()
     @click.argument("name")
     @click.argument("namespace")
-    @click.argument("user")
-    def create(name, namespace, user):  # pylint: disable=W0612
+    @click.argument("username")
+    def create(name, namespace, username):  # pylint: disable=W0612
         """Create blank project"""
         workspace = current_app.ws_handler.get_by_name(namespace)
+        if not workspace:
+            print("ERROR: Workspace not found")
+            return
+        user = User.query.filter_by(username=username).first()
+        if not user:
+            print("ERROR: User not found")
+            return
+        p = Project.query.filter_by(name=name, workspace_id=workspace.id).first()
+        if p:
+            print("ERROR: Project name already exists")
+            return
         project_params = dict(
             creator=user,
             name=name,
@@ -50,7 +62,7 @@ def create(name, namespace, user):  # pylint: disable=W0612
 
     @project.command()
     @click.argument("project-name")
-    @click.option("--version", required=True)
+    @click.option("--version", type=int, required=True)
     @click.option("--directory", type=click.Path(), required=True)
     def download(project_name, version, directory):  # pylint: disable=W0612
         """Download files for project at particular version"""

server/mergin/sync/db_events.py

@@ -6,7 +6,7 @@
 from flask import current_app, abort
 from sqlalchemy import event
 
-from .. import db
+from ..app import db
 
 
 def check(session):

server/mergin/sync/files.py

@@ -8,8 +8,7 @@
 from marshmallow import fields, EXCLUDE, pre_load, post_load, post_dump
 from pathvalidate import sanitize_filename
 
-from .. import ma
-from ..app import DateTimeWithZ
+from ..app import DateTimeWithZ, ma
 
 
 def mergin_secure_filename(filename: str) -> str:

server/mergin/sync/models.py

@@ -29,7 +29,7 @@
     ProjectFile,
 )
 from .storages.disk import move_to_tmp
-from .. import db
+from ..app import db
 from .storages import DiskStorage
 from .utils import is_versioned_file, is_qgis
 
@@ -100,15 +100,15 @@ def workspace(self):
         return project_workspace
 
     def cache_latest_files(self) -> None:
-        """Get project files from changes (FileHistory) and saved them for later use"""
+        """Get project files from changes (FileHistory) and save them for later use."""
         if self.latest_version is None:
             return
 
         query = f"""
             WITH latest_changes AS (
                 SELECT
                     fp.id,
-                    fp.project_id,
+                    pv.project_id,
                     max(pv.name) AS version
                 FROM
                     project_version pv
@@ -118,14 +118,13 @@ def cache_latest_files(self) -> None:
                     pv.project_id = :project_id
                     AND pv.name <= :latest_version
                 GROUP BY
-                    fp.id, fp.project_id
+                    fp.id, pv.project_id
             ), aggregates AS (
                 SELECT
                     project_id,
-                    array_agg(fh.id) AS files_ids
+                    COALESCE(array_agg(fh.id) FILTER (WHERE fh.change != 'delete'), ARRAY[]::INTEGER[]) AS files_ids
                 FROM latest_changes ch
                 LEFT OUTER JOIN file_history fh ON (fh.file_path_id = ch.id AND fh.project_version_name = ch.version)
-                WHERE fh.change != 'delete'
                 GROUP BY project_id
             )
             UPDATE latest_project_files pf

server/mergin/sync/private_api_controller.py

@@ -8,7 +8,7 @@
 from sqlalchemy.orm import defer
 from sqlalchemy import text, and_, desc, asc
 
-from .. import db
+from ..app import db
 from ..auth import auth_required
 from ..auth.models import User, UserProfile
 from .forms import AccessPermissionForm

server/mergin/sync/public_api_controller.py

@@ -32,7 +32,7 @@
 from gevent import sleep
 import base64
 from werkzeug.exceptions import HTTPException
-from .. import db
+from ..app import db
 from ..auth import auth_required
 from ..auth.models import User
 from .models import (
@@ -86,7 +86,8 @@
 from .errors import StorageLimitHit
 from ..utils import format_time_delta
 
-push_triggered = signal("push_triggered")
+push_finished = signal("push_finished")
+# TODO: Move to database events to handle all commits to project versions
 project_version_created = signal("project_version_created")
 
 
@@ -732,7 +733,6 @@ def project_push(namespace, project_name):
     if not ws:
         abort(404)
 
-    push_triggered.send(project)
     # fixme use get_latest
     pv = ProjectVersion.query.filter_by(
         project_id=project.id, name=project.latest_version
@@ -874,6 +874,7 @@ def project_push(namespace, project_name):
                 f"Transaction id: {upload.id}. No upload."
             )
             project_version_created.send(pv)
+            push_finished.send(pv)
             return jsonify(ProjectSchema().dump(project)), 200
         except IntegrityError as err:
             db.session.rollback()
@@ -1084,6 +1085,7 @@ def push_finish(transaction_id):
             f"Push finished for project: {project.id}, project version: {v_next_version}, transaction id: {transaction_id}."
         )
         project_version_created.send(pv)
+        push_finished.send(pv)
     except (psycopg2.Error, FileNotFoundError, DataSyncError, IntegrityError) as err:
         db.session.rollback()
         logging.exception(