updated code to latest version

code/app.py

@@ -1,38 +1,51 @@
-from flask import Flask, request, jsonify
+import uvicorn
-import os
+from fastapi import FastAPI, Request
 from dotenv import load_dotenv
-from ipaddress import ip_address, ip_network
 import logging
+from pydantic import BaseModel
+from typing import List
 
-load_dotenv()
+from utils import clone_repository
-app = Flask(__name__)
 
-@app.before_request
+class GiteaRequest(BaseModel):
-def check_authorized():
+    ref: str
-    """
+    before: str
-    Ensure request from allowed ip.
+    after: str
-    """
+    compare_url: str
-    # prepare variables
+    commits: List[dict]
-    assert 'ALLOWED_IP_RANGE' in os.environ
+    total_commits: int
-    allowed_ip_range = ip_network(os.getenv['ALLOWED_IP_RANGE'])
+    head_commit: dict
-    requesting_ip = ip_address(request.remote_addr)
+    repository: dict
-    # check that requesting ip is allowed
+    pusher: dict
-    if requesting_ip not in allowed_ip_range:
+    sender: dict
-        logging.info(f'received request from unauthorised ip: {request.remote_addr}')
-        return jsonify(status='forbidden'), 403
-    logging.info(f'received request from {request.remote_addr}')
 
-@app.before_request
+app = FastAPI()
-def check_media_type():
-    """
-    Ensure correct Content-Type of request
-    """
-    if not request.headers.get('Content-Type').lower().startswith('application/json'):
-        logging.error(f'received request with wrong content type')
-        return jsonify(status='unsupported media type'), 415
 
-@app.route('/test', methods=['POST'])
+@app.get('/')
-def test():
+async def root():
-    logging.debug(request.get_json(force=True))
+    return '<h1>Welcome to Gitea Runsner</h1>'
-    return jsonify(status='success', sender=request.remote_addr)
 
+@app.post('/test')
+async def test(request: Request):
+    """ Test connection to server. """
+    logging.debug('Content-Type: ', request.headers.get('content-type'))
+    return {'status': 'success'}
+
+@app.post('/event')
+def handle_event(request: GiteaRequest):
+    """ Handle event from webhook. """
+    # stop early if allowed
+    commit_msg = request.head_commit['message']
+    if commit_msg.lower().startswith('[skip ci]'):
+        return {'status': 'received task'}
+    # clone repository
+    repo_url = request.repository['clone_url']
+    local_repo_dir = clone_repository(repo_url)
+    print('local_repo_dir: ', local_repo_dir)
+    return {'status': 'success'}
+
+if __name__ == '__main__':
+    uvicorn.run(
+        app=app,
+        host='0.0.0.0'
+    )