see /docs for more information
""" return HTMLResponse(content=content_str, status_code=200) @app.post('/test') async def test(request: Request): """ Test connection to server. """ logging.debug('Content-Type: ', request.headers.get('content-type')) raise HTTPException( status_code=200, detail='test completed' ) @app.post('/event') def handle_event(request: GiteaRequest): """ Handle event from webhook. """ # extract information commit_msg = request.head_commit['message'] repo_url = request.repository['clone_url'] branch_name = request.ref.split('/')[-1] # stop early if asked if commit_msg.lower().startswith('[skip ci]'): raise HTTPException( status_code=200, detail='skipping ci' ) # clone repository local_repo_dir = clone_repository(repo_url) # locate runner script runner_script_path = local_repo_dir / 'runner_script.yml' if not runner_script_path.exists(): raise HTTPException( status_code=200, detail="no runner_script.yml in repository root" ) # load runner script runner_script = load_runner_script(path=runner_script_path) # determine pipeline to run if branch_name not in runner_script.pipelines.keys(): # get default pipeline if 'default' not in runner_script.pipelines.keys(): raise HTTPException( status_code=404, detail='no pipelines defined in runner_script.yml' ) pipeline = runner_script.pipelines['default'] else: # get custom pipeline pipeline = runner_script.pipelines[branch_name] logging.info(f'running pipeline {pipeline.name}') # 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, detail='received task' ) # consider to include link to page that show progress if __name__ == '__main__': load_dotenv('dev.env') setup_logging() listen_ip = os.getenv('LISTEN_IP', default='0.0.0.0') listen_port = int(os.getenv('LISTEN_PORT', default=8000)) logging.info(f'starting FastAPI server ({listen_ip}:{listen_port})') uvicorn.run( app=app, host=listen_ip, port=listen_port )