Code improvements and support for more output formats

.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,6 @@ build
 newrelic.ini
 newrelic.ini_backup
 exports
-nohup.out
+postgres-data
+*.out
+*.log

API/api_worker.py

@@ -0,0 +1,121 @@
+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, allow_bind_zip_filter
+
+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, params):
+    try:
+        start_time = dt.now()
+        bind_zip=params.bind_zip if allow_bind_zip_filter else True
+        # unique id for zip file and geojson for each export
+        params.output_type = params.output_type if params.output_type else RawDataOutputType.GEOJSON.value
+        params.file_name=format_file_name_str(params.file_name) if params.file_name else 'Galaxy_export'
+        exportname = f"{params.file_name}_{str(self.request.id)}_{params.output_type}"
+
+        logging.info("Request %s received", exportname)
+
+        geom_area, working_dir = RawData(params).extract_current_data(exportname)
+        inside_file_size = 0
+        if bind_zip:
+            logging.debug('Zip Binding Started !')
+            # saving file in temp directory instead of memory so that zipping file will not eat memory
+            upload_file_path = os.path.join(working_dir,os.pardir,f"{exportname}.zip")
+
+            zf = zipfile.ZipFile(upload_file_path, "w", zipfile.ZIP_DEFLATED)
+            for file_path in pathlib.Path(working_dir).iterdir():
+                zf.write(file_path, arcname=file_path.name)
+                inside_file_size += os.path.getsize(file_path)
+
+            # Compressing geojson file
+            zf.writestr("clipping_boundary.geojson",
+                        orjson.dumps(dict(params.geometry)))
+
+            zf.close()
+            logging.debug('Zip Binding Done !')
+        else:
+            for file_path in pathlib.Path(working_dir).iterdir():
+                upload_file_path=file_path
+                inside_file_size += os.path.getsize(file_path)
+                break # only take one file inside dir , if contains many it should be inside zip
+        # 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(upload_file_path, exportname, file_suffix='zip' if bind_zip else params.output_type.lower())
+        else:
+            download_url = str(upload_file_path) # give the static file download url back to user served from fastapi static export path
+
+        # getting file size of zip , units are in bytes converted to mb in response
+        zip_file_size = os.path.getsize(upload_file_path)
+        # watches the status code of the link provided and deletes the file if it is 200
+        if use_s3_to_upload:
+            watch_s3_upload(download_url, upload_file_path)
+        if use_s3_to_upload or bind_zip:
+            #remove working dir from the machine , if its inside zip / uploaded we no longer need it
+            remove_file(working_dir)
+        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": params.file_name, "process_time": response_time_str, "query_area": f"{round(geom_area,2)} 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

@@ -33,12 +33,14 @@
 from .hashtag_stats import router as hashtag_router
 from .tasking_manager import router as tm_router
 from .raw_data import router as raw_data_router
-from .download_export import router as download_router
-# from .test_router import router as test_router
+from .tasks import router as tasks_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, export_path, 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
+from fastapi.staticfiles import StaticFiles
 
 # only use sentry if it is specified in config blocks
 if config.get("SENTRY", "dsn", fallback=None):
@@ -50,12 +52,15 @@
         traces_sample_rate=config.get("SENTRY", "rate")
     )
 
+
 run_env = config.get("API_CONFIG", "env", fallback='prod')
 if run_env.lower() == 'dev':
     # This is used for local setup for auth login
     import os
     os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
 
+
+
 app = FastAPI(title="Galaxy API")
 
 # app.include_router(test_router)
@@ -71,15 +76,20 @@
 app.include_router(tm_router)
 app.include_router(status_router)
 app.include_router(raw_data_router)
+app.include_router(tasks_router)
+
 
-if use_s3_to_upload is False:
-    # only mount the disk if config is set to disk
-    app.include_router(download_router)
 
 
 app = VersionedFastAPI(app, enable_latest=True,
                        version_format='{major}', prefix_format='/v{major}')
 
+if use_s3_to_upload is False:
+    # only mount the disk if config is set to disk
+    app.mount("/exports", StaticFiles(directory=export_path), name="exports")
+
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
 origins = ["*"]