Merge pull request 'added data queue' (#9) from main into release

Reviewed-on: http://192.168.1.2:3000/brian/bandwidth_probing/pulls/9

.gitea/workflows/default.yaml

@@ -0,0 +1,28 @@
+name: pipeline
+run-name: ${{ gitea.actor }} is running script
+runs-on: ubuntu-latest
+on:
+  push:
+    branches-ignore:
+      - "release"
+image: python:3.10
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.10
+        uses: https://github.com/actions/setup-python@v3
+        with:
+          python-version: "3.10"
+          architecture: "x64"
+      - name: Install Packages
+        run: |
+          python -m pip install --upgrade pip
+          pip install flake8 mypy pytest
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: PEP8 check
+        run: flake8 ./code --benchmark --exit-zero
+      - name: Type check
+        run: mypy ./code

.gitea/workflows/pipeline.yaml

@@ -1,26 +0,0 @@
-name: Python application
-run-name: ${{ gitea.actor }} is building the application
-on: [push]
-image: python:3.10
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python 3.10
-      uses: https://github.com/actions/setup-python@v3
-      with:
-        python-version: "3.10"
-        architecture: "x64"
-    - name: Install Dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install flake8 pytest
-        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
-    - name: Lint with flake8
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 ./code
-        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        # flake8 ./code --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics

.gitea/workflows/release.yaml

@@ -0,0 +1,52 @@
+name: Docker
+run-name: Build and Publish Service
+runs-on: ubuntu-latest
+on:
+  push:
+    tags:
+      - 'v[0-9]+.[0-9]+.[0-9]+'
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.10
+        uses: https://github.com/actions/setup-python@v3
+        with:
+          python-version: "3.10"
+          architecture: "x64"
+      - name: Install Packages
+        run: |
+          python -m pip install --upgrade pip
+          pip install flake8 mypy pytest
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: PEP8 check
+        run: flake8 ./code --benchmark --exit-zero
+      - name: Type check
+        run: mypy ./code
+  publish:
+    name: Publish
+    runs-on: ubuntu-latest
+    needs: [test]
+    steps:
+      - uses: actions/checkout@v3
+      -
+        name: Set Environment Variables
+        id: vars
+        env:
+          repository: ${{gitea.repository}}
+          ref_name: ${{gitea.ref_name}}
+        run: |
+          echo $repository
+          echo ::set-output name=OWNER::${repository#/*}
+          echo ::set-output name=PROJECT::${repository#*/}
+          echo $ref_name
+          echo ::set-output name=TAG::${ref_name#refs/*/}
+      -
+        name: Push Image
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          push: true
+          tags: 192.168.1.2:5000/${{gitea.repository}}:${{steps.vars.outputs.TAG}}

code/bandwidth.py

@@ -1,6 +1,6 @@
 from speedtest import Speedtest
 from dotenv import load_dotenv
-from typing import List
+import logging
 
 
 def measure():
@@ -13,6 +13,7 @@ def measure():
     s.download(threads=None)
     s.upload(threads=None)
     s.results.share()
+    logging.debug('finished')
     return s.results.dict()
 
 

code/database.py

@@ -1,27 +1,31 @@
 from pymongo import MongoClient
 from dotenv import load_dotenv
+from configparser import ConfigParser
+from pathlib import Path
 import logging
-import os
+
+
+# load default values
+config = ConfigParser()
+config.read(Path(__file__).parent / 'defaults.ini')
+DB_IP_ADDRESS = config.get('database', 'db_ip_address')
+DB_NAME = config.get('database', 'db_name')
+DB_COLLECTION_NAME = config.get('database', 'db_collection_name')
+
 
 def connect(
-    ip_addr: str | None = None,
+    ip_addr: str = DB_IP_ADDRESS,
-    db_name: str | None = None,
+    db_name: str = DB_NAME,
-    collection_name: str | None = None
+    collection_name: str = DB_COLLECTION_NAME
 ):
     """
     Connect to MongoDB database and return collection.
     """
-    # ensure variables are set
-    if ip_addr is None:
-        ip_addr = os.getenv('DB_IP_ADDRESS')
-    if db_name is None:
-        db_name = os.getenv('DB_NAME')
-    if collection_name is None:
-        collection_name = os.getenv('DB_COLLECTION_NAME')
     # connect to database
-    client = MongoClient(ip_addr)
+    client: MongoClient = MongoClient(ip_addr)
     db = client[db_name]
     collection = db[collection_name]
+    logging.debug('finished')
     return collection
 
 

code/defaults.ini

@@ -0,0 +1,15 @@
+[main]
+logger_level=info
+user_id=0
+replicates=3
+trigger_interval_seconds=60
+
+[database]
+db_ip_address=192.168.1.2
+db_name=bandwidth_probing
+db_collection_name=data
+
+[discord]
+webhook_url=https://discord.com/api/webhooks/1127970367047225354/wNkoRex4OncMw11OCJg6atR_xgHS2VhLrku4jYIUw81Kr4sutdD3Tt-XAltG0UQy14rZ
+service_name=bandwidth_probing
+logger_level=warning

code/initialise_app.py

@@ -1,53 +1,63 @@
 from discord_logging.handler import DiscordHandler
 from dotenv import load_dotenv
+from configparser import ConfigParser
+from pathlib import Path
 import logging
 import os
 
 
-def initialise_app():
+# load default values
+config = ConfigParser()
+config.read(Path(__file__).parent / 'defaults.ini')
+LOGGER_LEVEL = config.get('main', 'logger_level')
+DISCORD_SERVICE_NAME = config.get('discord', 'service_name')
+DISCORD_WEBHOOK_URL = config.get('discord', 'webhook_url')
+DISCORD_LOGGER_LEVEL = config.get('discord', 'logger_level')
+
+
+def initialise_app(
+    logger_level: str = LOGGER_LEVEL,
+    discord_service_name: str = DISCORD_SERVICE_NAME,
+    discord_webhook_url: str = DISCORD_WEBHOOK_URL,
+    discord_logger_level: str = DISCORD_LOGGER_LEVEL
+):
     """
-    Convienience function that ensures eveything is ready before running main loop.
+    Convienience function that ensures eveything is ready
+    before running the main loop.
     """
-    # load environment variables from file
+    # load environment variables
     load_dotenv()
+    logger_level = os.getenv(
+        'LOGGER_LEVEL',
+        default=LOGGER_LEVEL
+    )
+    discord_service_name = os.getenv(
+        'DISCORD_SERVICE_NAME',
+        default=DISCORD_SERVICE_NAME
+    )
+    discord_webhook_url = os.getenv(
+        'DISCORD_WEBHOOK_URL',
+        default=DISCORD_WEBHOOK_URL
+    )
+    discord_logger_level = os.getenv(
+        'DISCORD_LOGGER_LEVEL',
+        default=DISCORD_LOGGER_LEVEL
+    )
     # setup logging
     fmt = '|%(asctime)s| %(levelname)s %(filename)s:%(funcName)s:%(message)s'
     datefmt = '%Y-%m-%d %H:%M:%S'
-    logger_level = os.getenv('LOGGER_LEVEL', default=None)
-    if logger_level is None:
-        logger_level = 'info'
-        logging.info(f'environment variable LOGGER_LEVEL not set. Using default (info)')
     level = getattr(logging, logger_level.upper())
     logging.basicConfig(format=fmt, datefmt=datefmt, level=level)
     # setup logging to discord
-    discord_url = os.getenv('DISCORD_URL', default=None)
-    assert discord_url is not None
     logger = logging.getLogger()
     discord_handler = DiscordHandler(
-        service_name = 'bandwidth_probing',
+        service_name=discord_service_name,
-        webhook_url = discord_url,
+        webhook_url=discord_webhook_url,
     )
     discord_handler.setFormatter(logging.Formatter('%(message)s'))
-    discord_handler.setLevel(logging.WARNING)
+    level = getattr(logging, discord_logger_level.upper())
+    discord_handler.setLevel(level=level)
     logger.addHandler(discord_handler)
-    # ensure environment variables are set
-    necessary_list = [
-        'DB_IP_ADDRESS',
-        'DB_NAME',
-        'DB_COLLECTION_NAME'
-    ]
-    for var in necessary_list:
-        assert var in os.environ, f'environment variable {var} not set.'
-    # set defaults if not yet handled
-    default_dict = {
-        'USER_ID': 0,
-        'REPLICATES': 3,
-        'TRIGGER_INTERVAL_SECONDS': 60
-    }
-    for k, v in default_dict.items():
-        if not k in os.environ:
-            logging.info(f'environment variable {k} not set. Using default ({v}).')
-            os.environ[k] = str(v)
 
 
 if __name__ == '__main__':

code/main.py

@@ -1,51 +1,103 @@
 import time
 import logging
-import time
 import os
+import json
+from configparser import ConfigParser
+from pathlib import Path
+from queue import Queue
 from initialise_app import initialise_app
 from bandwidth import measure as measure_bandwidth
 from database import connect
 
-def event():
+
-    logging.info('started event')
+# load default values
-    # get env vars
+config = ConfigParser()
-    replicates = int(os.getenv('REPLICATES'))
+config.read(Path(__file__).parent / 'defaults.ini')
-    # setup database connection
+REPLICATES = int(config.get('main', 'replicates'))
-    db = connect()
+USER_ID = int(config.get('main', 'user_id'))
+TRIGGER_INTERVAL_SECONDS = float(
+    config.get(
+        'main',
+        'trigger_interval_seconds'
+    )
+)
+
+
+def event(
+    replicates: int = REPLICATES,
+    user_id: int = USER_ID
+):
+    logging.debug('started event')
+    # load environment variables
+    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
+    user_id = int(os.getenv('USER_ID', default=USER_ID))
+    # setup data queue
+    data_queue: Queue = Queue()
     # run event
     for rep_num in range(replicates):
-        logging.info(f'running replicate {rep_num}')
+        logging.debug(f'running replicate {rep_num+1} of {replicates}')
         # do measurement
         try:
-            res = measure_bandwidth()
+            data = measure_bandwidth()
         except Exception as e:
             logging.error(f'failed to measure bandwidth: {e}')
             continue
-        # upload result to database
+        # add data to queue
+        data_queue.put(data)
+        # polite pause
+        time.sleep(1)
+    # check if queue should be flushed
+    queue_size = data_queue.qsize()
+    if queue_size == 0:
+        logging.debug('queue empty, continuing')
+        return
+    # flush queue to database
+    db = connect()
+    for i in range(queue_size):
+        # get data
+        data = data_queue.get()
+        # prepare payload
         payload_dict = {
-            'user_id': os.getenv('USER_ID'),
+            'user_id': user_id,
-            'data': res
+            'data': data
         }
+        # send to database
         try:
             db_id = db.insert_one(payload_dict).inserted_id
         except Exception as e:
-            logging.error(f'failed sending results to database: {e}')
+            logging.error('failed sending data to database')
+            logging.info(
+                'failed pushing to database:\n'
+                f'{json.dumps(payload_dict, indent=4)}\n'
+                f'with error: {e}'
+            )
+            data_queue.put(data)    # put data back in queue
             continue
-        logging.debug(f'data sent to database received db_id: {db_id}')
+        else:
-        time.sleep(1)
+            logging.debug(f'data sent to database received id: {db_id}')
-    logging.info('finished event')
+    logging.debug('finished')
+
 
 if __name__ == '__main__':
     initialise_app()
     trigger_time = time.time()
     while True:
         if time.time() >= trigger_time:
+            logging.debug('triggered event')
             # set new trigger time
-            trigger_time += int(os.getenv('TRIGGER_INTERVAL_SECONDS'))
+            trigger_interval = float(
+                os.getenv(
+                    'TRIGGER_INTERVAL_SECONDS',
+                    default=TRIGGER_INTERVAL_SECONDS
+                )
+            )
+            trigger_time += trigger_interval
             # run event
             try:
                 event()
             except Exception as e:
                 logging.error(e)
+            else:
+                logging.info('finished event')
+        # polite pause
         time.sleep(1)
-

mypy.ini

@@ -0,0 +1,7 @@
+[mypy]
+
+[mypy-discord_logging.*]
+ignore_missing_imports = True
+
+[mypy-speedtest.*]
+ignore_missing_imports = True