updated ignored modules

.gitea/gpu_runner/script.yml

@@ -1,3 +1,4 @@
 script: |
   echo "start of script"
+  sleep 15
   echo "end of script"

code/app.py

@@ -1,18 +1,23 @@
 import uvicorn
-from fastapi import FastAPI, Request, HTTPException
+from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
 from fastapi.responses import HTMLResponse
 from dotenv import load_dotenv
 import logging
 import os
-import shutil
+import httpx
 
-from utils import clone_repository, execute_runner_script
+from utils import clone_repository, execute_script
 from setup_logging import setup_logging
 from gitea_request import GiteaRequest
 
 app = FastAPI()
 
 
+async def post_callback(url: str, payload: dict):
+    async with httpx.AsyncClient() as client:
+        await client.post(url, json=payload)
+
+
 @app.get('/', response_class=HTMLResponse)
 async def root():
     content_str = """
@@ -40,7 +45,7 @@ async def test(request: Request):
 
 
 @app.post('/')
-def handle_event(request: GiteaRequest):
+def handle_event(request: GiteaRequest, background_tasks: BackgroundTasks):
     """ Handle event from webhook. """
     # extract information
     commit_msg = request.head_commit['message']
@@ -54,48 +59,23 @@ def handle_event(request: GiteaRequest):
         )
     # clone repository
     local_repo_dir = clone_repository(repo_url)
-    # locate runner script
-    runner_script_path = (
-        local_repo_dir / 
-        '.gitea' / 
-        'gpu_runner' / 
-        'script.yml'
+    # # locate runner script
+    # runner_script_path = (
+    #     local_repo_dir /
+    #     '.gitea' /
+    #     'gpu_runner' /
+    #     'script.yml'
+    # )
+    # if not runner_script_path.exists():
+    #     raise HTTPException(
+    #         status_code=200,
+    #         detail=f"{runner_script_path} does not exist"
+    #     )
+    # queue runner script
+    background_tasks.add_task(
+        execute_script,
+        local_repo_dir
     )
-    if not runner_script_path.exists():
-        raise HTTPException(
-            status_code=200,
-            detail="no runner_script.yml in repository root"
-        )
-    venv_dir = local_repo_dir / 'venv'
-    try:
-        # create virtualenv
-        os.system(f'virtualenv -p python3.10 {venv_dir}')
-        logging.info('created venv')
-        os.system(f'source {venv_dir}/bin/activate')
-        logging.info('activated venv')
-        # execute runner script
-        execute_runner_script(path=runner_script_path)
-    except:
-        logging.error('failed running script')
-    finally:
-        os.system('deactivate')
-        logging.info('deactivated venv')
-        shutil.rmtree(venv_dir)
-        logging.info('removed venv')
-    # # TODO: create virtualenv?
-    # # install requirements
-    # assert pipeline.requirements is not None
-    # requirements_path = local_repo_dir / pipeline.requirements
-    # os.system(f'pip install -r {requirements_path}')
-    # for cmd in pipeline.script:
-    #     try:
-    #         os.system(cmd)
-    #     except Exception as e:
-    #         raise HTTPException(
-    #             status_code=503,
-    #             detail=f'failed running {cmd}: {e}'
-    #         )
-
     # send response
     raise HTTPException(
         status_code=202,
@@ -113,5 +93,6 @@ if __name__ == '__main__':
     uvicorn.run(
         app=app,
         host=listen_ip,
-        port=listen_port
+        port=listen_port,
+        workers=1
     )

code/main.py

@@ -7,7 +7,8 @@ import os
 from configparser import ConfigParser
 from pathlib import Path
 import socket
-
+# from multiprocessing import Queue
+# from worker import Worker
 
 # load default values
 config = ConfigParser()
@@ -23,9 +24,17 @@ if __name__ == '__main__':
     listen_ip = os.getenv('LISTEN_IP', default=LISTEN_IP)
     listen_port = int(os.getenv('LISTEN_PORT', default=LISTEN_PORT))
     own_ip = socket.gethostbyname(socket.gethostname())
+    # # setup work queue
+    # work_queue: Queue = Queue()
+    # # setup worker process
+    # worker = Worker(queue=work_queue)
+    # start server
     logging.info(f'starting FastAPI server ({own_ip}:{listen_port})')
+    # try:
     uvicorn.run(
         app=app,
         host=listen_ip,
         port=listen_port
     )
+    # except KeyboardInterrupt:
+    #     worker.stop()

code/utils.py

@@ -31,9 +31,40 @@ def clone_repository(
     return dst_dir
 
 
-def execute_runner_script(path: Path) -> None:
-    rs = RunnerScript.from_file(path)
+def execute_script(
+    local_repo_dir: Path
+) -> None:
+    # locate the script
+    script_path = (
+        local_repo_dir /
+        '.gitea' /
+        'gpu_runner' /
+        'script.yml'
+    )
+    # determine original working dir
+    org_dir = os.getcwd()
+    # initialise command list
+    cmd_list = [
+        f'cd {local_repo_dir.resolve()}',
+        'virtualenv -p python3.10 venv',
+        'source venv/bin/activate',
+    ]
+    # add in runner script
+    rs = RunnerScript.from_file(path=script_path)
     for cmd in rs.script:
-        print(f'> {cmd}')
-        os.system(cmd)
+        cmd_list.append(cmd)
+    # append cleanup commands
+    cmd_list.append('deactivate')
+    cmd_list.append(f'cd {org_dir}')
+    cmd_list.append(f'rm -rf {local_repo_dir.resolve()}')
+    # build command string
+    cmd_str = ' && '.join(cmd_list)
+    os.system(cmd_str)
     logging.debug('finished')
+
+
+if __name__ == '__main__':
+    local_repo_dir = Path(__file__).parent.parent / 'repo' / 'gpu_runner'
+    execute_script(
+        local_repo_dir=local_repo_dir
+    )

code/worker.py

@@ -0,0 +1,40 @@
+from multiprocessing import Process, Queue
+import time
+from typing import List
+import os
+
+
+class Worker(object):
+    def __init__(self, queue: Queue):
+        self.run_flag = True
+        self.queue = queue
+        self.process = Process(
+            target=self.handle_jobs,
+            args=(self.queue,)
+        )
+        time.sleep(1)
+        self.process.start()
+
+    def stop(self):
+        self.queue.put(['exit'])
+        self.process.join()
+
+    @staticmethod
+    def handle_jobs(queue: Queue):
+        while True:
+            if queue.empty():
+                time.sleep(1)
+                continue
+            cmd_list: List[str] = queue.get()
+            print('received job:\n', cmd_list)
+            if cmd_list[0].lower() == 'exit':
+                break
+            cmd_str = ' && '.join(cmd_list)
+            os.system(cmd_str)
+
+
+if __name__ == '__main__':
+    work_queue: Queue = Queue()
+    worker = Worker(queue=work_queue)
+    work_queue.put(['echo "hello world"', 'echo "hello world again"'])
+    worker.stop()

mypy.ini

@@ -10,4 +10,7 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 
 [mypy-fastapi.*]
+ignore_missing_imports = True
+
+[mypy-httpx.*]
 ignore_missing_imports = True