[Feature] Task Queue System & Rate Limit on API

.github/workflows/Unit-Test.yml

@@ -13,17 +13,22 @@ jobs:
   deploy:
     runs-on:
       ubuntu-latest
-
+    services:
+     postgres:
+      image: postgis/postgis:14-3.3
+      env:
+        POSTGRES_PASSWORD: admin
+        POSTGRES_DB: insights
+      ports:
+        - 5434:5432
+      options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 2
     steps:
       - uses: actions/checkout@v2
       - name: Set up Python 3.8
         uses: actions/setup-python@v1
         with:
           python-version: 3.8
-      - name: Check postgresql version 
-        run: |
-          psql -V
-      - name: Remove postgresql version  14
+      - name: Clean up PSQL
         run: |
           sudo apt-get --purge remove postgresql
           sudo apt-get purge postgresql*
@@ -36,17 +41,31 @@ jobs:
         run: |
           sudo apt-get update
           sudo apt install postgis postgresql-12-postgis-3
-      - name: Install gdal 
-        run: |
-          sudo apt-add-repository ppa:ubuntugis/ubuntugis-unstable
-          sudo apt-get update
-          sudo apt-get install gdal-bin libgdal-dev
-
+
+      - name: Create Databases
+        run : |
+          export PGPASSWORD='admin';
+          psql -U postgres -h localhost -p 5434 -c "CREATE DATABASE underpass;"
+          psql -U postgres -h localhost -p 5434 -c "CREATE DATABASE tm;"
+          psql -U postgres -h localhost -p 5434 -c "CREATE DATABASE raw;"
+
+      - name: Insert sample db data
+        run : |
+          export PGPASSWORD='admin';
+          psql -U postgres -h localhost -p 5434 insights < tests/src/fixtures/insights.sql
+          psql -U postgres -h localhost -p 5434 raw  < tests/src/fixtures/raw_data.sql
+          psql -U postgres -h localhost -p 5434 underpass < tests/src/fixtures/underpass.sql
+          wget https://raw.githubusercontent.com/hotosm/tasking-manager/develop/tests/database/tasking-manager.sql
+          psql -U postgres -h localhost -p 5434 tm < tasking-manager.sql
+
       - name: Install Dependencies
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements.txt
           pip install -e .
+      - name: Creating config.txt
+        run: |
+          mv src/config.txt.sample src/config.txt
       - name: Run Tests
         run: |
-          py.test -v -s
+          py.test -v -s

.github/workflows/build.yml

@@ -0,0 +1,86 @@
+name: Check Build
+on:
+  push:
+    branches:
+      - master
+      - develop
+  pull_request:
+    branches:
+      - master
+      - develop
+
+jobs:
+  build:
+    timeout-minutes: 4
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgis/postgis:14-3.3
+        env:
+          POSTGRES_PASSWORD: admin
+          POSTGRES_DB: insights
+        ports:
+          - 5434:5432
+        options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 2
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python 3.8
+        uses: actions/setup-python@v1
+        with:
+          python-version: 3.8
+
+      - name: Create Databases
+        run: |
+          export PGPASSWORD='admin';
+          psql -U postgres -h localhost -p 5434 -c "CREATE DATABASE underpass;"
+          psql -U postgres -h localhost -p 5434 -c "CREATE DATABASE tm;"
+          psql -U postgres -h localhost -p 5434 -c "CREATE DATABASE raw;"
+
+      - name: Insert sample db data
+        run: |
+          export PGPASSWORD='admin';
+          psql -U postgres -h localhost -p 5434 insights < tests/src/fixtures/insights.sql
+          psql -U postgres -h localhost -p 5434 insights < tests/src/fixtures/mapathon_summary.sql
+          psql -U postgres -h localhost -p 5434 raw  < tests/src/fixtures/raw_data.sql
+          psql -U postgres -h localhost -p 5434 underpass < tests/src/fixtures/underpass.sql
+          wget https://raw.githubusercontent.com/hotosm/tasking-manager/develop/tests/database/tasking-manager.sql
+          psql -U postgres -h localhost -p 5434 tm < tasking-manager.sql
+
+      - name: Install gdal
+        run: |
+          sudo apt-get -y install gdal-bin python3-gdal && sudo apt-get -y autoremove && sudo apt-get clean
+
+      - name: Install redis
+        run: |
+          sudo apt install lsb-release
+          curl -fsSL https://packages.redis.io/gpg | sudo gpg --dearmor -o /usr/share/keyrings/redis-archive-keyring.gpg
+          echo "deb [signed-by=/usr/share/keyrings/redis-archive-keyring.gpg] https://packages.redis.io/deb $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/redis.list
+          sudo apt-get update
+          sudo apt-get install redis
+          redis-cli ping
+
+      - name: Install Dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install -e .
+      - name: Creating config.txt
+        run: |
+          mv src/config.txt.sample src/config.txt
+      - name: Run uvicorn server
+        run: |
+          uvicorn API.main:app &
+        env:
+          PORT: 8000
+      - name: Run celery server
+        run: |
+          celery --app API.api_worker worker --loglevel=INFO &
+      - name: Run flower dashboard
+        run: |
+          celery --app API.api_worker flower --port=5555 --broker=redis://localhost:6379/ &
+      - name: Run mapathon summary endpoint
+        run: |
+          curl -d '{"project_ids": [11224, 10042, 9906, 1381, 11203, 10681, 8055, 8732, 11193, 7305,11210, 10985, 10988, 11190, 6658, 5644, 10913, 6495, 4229],"fromTimestamp":"2021-08-27T9:00:00","toTimestamp":"2021-08-27T11:00:00","hashtags": ["mapandchathour2021"]}' -H 'Content-Type: application/json' http://127.0.0.1:8000/v1/mapathon/summary/
+      - name: Run rawdata current snapshot
+        run: |
+          curl -d '{"geometry":{"type":"Polygon","coordinates":[[[83.96919250488281,28.194446860487773],[83.99751663208006,28.194446860487773],[83.99751663208006,28.214869548073377],[83.96919250488281,28.214869548073377],[83.96919250488281,28.194446860487773]]]}}' -H 'Content-Type: application/json' http://127.0.0.1:8000/v2/raw-data/current-snapshot/

.gitignore

@@ -12,4 +12,5 @@ build
 newrelic.ini
 newrelic.ini_backup
 exports
-nohup.out
+*.out
+*.log

API/api_worker.py

@@ -0,0 +1,144 @@
+import os
+import pathlib
+import orjson
+import shutil
+import time
+import requests
+from datetime import datetime as dt
+import zipfile
+from celery import Celery
+from src.galaxy.config import config
+from fastapi.responses import JSONResponse
+from src.galaxy.query_builder.builder import format_file_name_str
+from src.galaxy.validation.models import RawDataOutputType
+from src.galaxy.app import RawData, S3FileTransfer
+from src.galaxy.config import use_s3_to_upload, logger as logging, config
+
+celery = Celery(__name__)
+celery.conf.broker_url = config.get(
+    "CELERY", "CELERY_BROKER_URL", fallback="redis://localhost:6379"
+)
+celery.conf.result_backend = config.get(
+    "CELERY", "CELERY_RESULT_BACKEND", fallback="redis://localhost:6379"
+)  # using redis as backend , make sure you have redis server started on your system on port 6379
+
+celery.conf.task_serializer = 'pickle'
+celery.conf.result_serializer = 'pickle'
+celery.conf.accept_content = ['application/json', 'application/x-python-serialize']
+
+
+@celery.task(bind=True, name="process_raw_data")
+def process_raw_data(self, incoming_scheme, incoming_host, params):
+    try:
+        start_time = dt.now()
+        if params.output_type is None:  # if no ouput type is supplied default is geojson output
+            params.output_type = RawDataOutputType.GEOJSON.value
+
+        # unique id for zip file and geojson for each export
+        if params.file_name:
+            # need to format string from space to _ because it is filename , may be we need to filter special character as well later on
+            formatted_file_name = format_file_name_str(params.file_name)
+            # exportname = f"{formatted_file_name}_{datetime.now().isoformat()}_{str(self.request.id)}"
+            exportname = f"""{formatted_file_name}_{str(self.request.id)}_{params.output_type}"""  # disabled date for now
+
+        else:
+            # exportname = f"Raw_Export_{datetime.now().isoformat()}_{str(self.request.id)}"
+            exportname = f"Raw_Export_{str(self.request.id)}_{params.output_type}"
+
+        logging.info("Request %s received", exportname)
+
+        dump_temp_file, geom_area, root_dir_file = RawData(
+            params).extract_current_data(exportname)
+        path = f"""{root_dir_file}{exportname}/"""
+
+        if os.path.exists(path) is False:
+            return JSONResponse(
+                status_code=400,
+                content={"Error": "Request went too big"}
+            )
+
+        logging.debug('Zip Binding Started !')
+        # saving file in temp directory instead of memory so that zipping file will not eat memory
+        zip_temp_path = f"""{root_dir_file}{exportname}.zip"""
+        zf = zipfile.ZipFile(zip_temp_path, "w", zipfile.ZIP_DEFLATED)
+
+        directory = pathlib.Path(path)
+        for file_path in directory.iterdir():
+            zf.write(file_path, arcname=file_path.name)
+
+        # Compressing geojson file
+        zf.writestr("clipping_boundary.geojson",
+                    orjson.dumps(dict(params.geometry)))
+
+        zf.close()
+        logging.debug('Zip Binding Done !')
+        inside_file_size = 0
+        for temp_file in dump_temp_file:
+            # clearing tmp geojson file since it is already dumped to zip file we don't need it anymore
+            if os.path.exists(temp_file):
+                inside_file_size += os.path.getsize(temp_file)
+
+        # remove the file that are just binded to zip file , we no longer need to store it
+        remove_file(path)
+        # check if download url will be generated from s3 or not from config
+        if use_s3_to_upload:
+            file_transfer_obj = S3FileTransfer()
+            download_url = file_transfer_obj.upload(zip_temp_path, exportname)
+        else:
+            # getting from config in case api and frontend is not hosted on same url
+            client_host = config.get(
+                "API_CONFIG", "api_host", fallback=f"""{incoming_scheme}://{incoming_host}""")
+            client_port = config.get("API_CONFIG", "api_port", fallback=8000)
+
+            if client_port:
+                download_url = f"""{client_host}:{client_port}/v1/exports/{exportname}.zip"""  # disconnected download portion from this endpoint because when there will be multiple hits at a same time we don't want function to get stuck waiting for user to download the file and deliver the response , we want to reduce waiting time and free function !
+            else:
+                download_url = f"""{client_host}/v1/exports/{exportname}.zip"""  # disconnected download portion from this endpoint because when there will be multiple hits at a same time we don't want function to get stuck waiting for user to download the file and deliver the response , we want to reduce waiting time and free function !
+
+        # getting file size of zip , units are in bytes converted to mb in response
+        zip_file_size = os.path.getsize(zip_temp_path)
+        # watches the status code of the link provided and deletes the file if it is 200
+        watch_s3_upload(download_url, zip_temp_path)
+        response_time = dt.now() - start_time
+        response_time_str = str(response_time)
+        logging.info(
+            f"Done Export : {exportname} of {round(inside_file_size/1000000)} MB / {geom_area} sqkm in {response_time_str}")
+        return {"download_url": download_url, "file_name": exportname, "process_time": response_time_str, "query_area": f"""{geom_area} Sq Km """, "binded_file_size": f"""{round(inside_file_size/1000000,2)} MB""", "zip_file_size_bytes": zip_file_size}
+
+    except Exception as ex:
+        raise ex
+
+
+def remove_file(path: str) -> None:
+    """Used for removing temp file dir and its all content after zip file is delivered to user"""
+    try:
+        shutil.rmtree(path)
+    except OSError as ex:
+        logging.error("Error: %s - %s.", ex.filename, ex.strerror)
+
+
+def watch_s3_upload(url: str, path: str) -> None:
+    """Watches upload of s3 either it is completed or not and removes the temp file after completion
+
+    Args:
+        url (_type_): url generated by the script where data will be available
+        path (_type_): path where temp file is located at
+    """
+    start_time = time.time()
+    remove_temp_file = True
+    check_call = requests.head(url).status_code
+    if check_call != 200:
+        logging.debug("Upload is not done yet waiting ...")
+        while check_call != 200:  # check until status is not green
+            check_call = requests.head(url).status_code
+            if time.time() - start_time > 300:
+                logging.error(
+                    "Upload time took more than 5 min , Killing watch : %s , URL : %s", path, url)
+                remove_temp_file = False  # don't remove the file if upload fails
+                break
+            time.sleep(3)  # check each 3 second
+    # once it is verfied file is uploaded finally remove the file
+    if remove_temp_file:
+        logging.debug(
+            "File is uploaded at %s , flushing out from %s", url, path)
+        os.unlink(path)

API/main.py

@@ -37,8 +37,11 @@
 # from .test_router import router as test_router
 from .status import router as status_router
 from src.galaxy.db_session import database_instance
-from src.galaxy.config import use_connection_pooling, use_s3_to_upload, logger as logging, config
+from src.galaxy.config import limiter, use_connection_pooling, use_s3_to_upload, logger as logging, config
 from fastapi_versioning import VersionedFastAPI
+from slowapi import _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+
 
 # only use sentry if it is specified in config blocks
 if config.get("SENTRY", "dsn", fallback=None):
@@ -56,6 +59,8 @@
     import os
     os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
 
+
+
 app = FastAPI(title="Galaxy API")
 
 # app.include_router(test_router)
@@ -81,6 +86,9 @@
                        version_format='{major}', prefix_format='/v{major}')
 
 
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
 origins = ["*"]