mypy compliant

code/app.py

@@ -5,7 +5,7 @@ from dotenv import load_dotenv
 import logging
 import os
 
-from utils import clone_repository, load_runner_script
+from utils import clone_repository, execute_runner_script
 from setup_logging import setup_logging
 from gitea_request import GiteaRequest
 
@@ -60,34 +60,23 @@ def handle_event(request: GiteaRequest):
             status_code=200,
             detail="no runner_script.yml in repository root"
         )
-    # load runner script
+    # execute runner script
-    runner_script = load_runner_script(path=runner_script_path)
+    logging.info('rexecuting runner script')
-    # determine pipeline to run
+    execute_runner_script(path=runner_script_path)
-    if branch_name not in runner_script.pipelines.keys():
+    
-        # get default pipeline
+    # # TODO: create virtualenv?
-        if 'default' not in runner_script.pipelines.keys():
+    # # install requirements
-            raise HTTPException(
+    # assert pipeline.requirements is not None
-                status_code=404,
+    # requirements_path = local_repo_dir / pipeline.requirements
-                detail='no pipelines defined in runner_script.yml'
+    # os.system(f'pip install -r {requirements_path}')
-            )
+    # for cmd in pipeline.script:
-        pipeline = runner_script.pipelines['default']
+    #     try:
-    else:
+    #         os.system(cmd)
-        # get custom pipeline
+    #     except Exception as e:
-        pipeline = runner_script.pipelines[branch_name]
+    #         raise HTTPException(
-    logging.info(f'running pipeline {pipeline.name}')
+    #             status_code=503,
-    # TODO: create virtualenv?
+    #             detail=f'failed running {cmd}: {e}'
-    # 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(

code/runner_script.py

@@ -15,7 +15,7 @@ class RunnerScript(BaseModel):
             content_dict = yaml.safe_load(fh)
         script_list = content_dict['script'].split('\n')
         logging.debug('finished')
-        return RunnerScript.model_validate({'script': script_list})
+        return RunnerScript.model_validate({'script': script_list})  # type: ignore
 
 
 if __name__ == '__main__':

code/setup_logging.py

@@ -12,7 +12,7 @@ LOGGER_LEVEL = config.get('main', 'logger_level')
 
 
 def setup_logging(
-    logger_level: Literal['debug', 'info', 'warning', 'error'] = LOGGER_LEVEL
+    logger_level: str = LOGGER_LEVEL
 ):
     fmt = (
         '%(asctime)s | '

code/utils.py

@@ -4,7 +4,6 @@ from git import Repo
 import logging
 import shutil
 import os
-import yaml
 from runner_script import RunnerScript
 
 
@@ -24,10 +23,8 @@ def clone_repository(repo_url: AnyHttpUrl) -> Path:
     return dst_dir
 
 
-def load_runner_script(path: Path) -> RunnerScript:
+def execute_runner_script(path: Path) -> None:
-    assert path.exists()
+    script_list = RunnerScript.from_file(path)
-    with open(path, 'r') as fh:
+    for cmd in script_list:
-        script = yaml.safe_load(fh)
+        os.system(cmd)
-    runner_script = RunnerScript.parse_obj(script)
     logging.debug('finished')
-    return runner_script