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] logging.info(f'branch: {branch_name}') # stop early if asked if commit_msg.lower().startswith('[skip ci]'): raise HTTPException( status_code=200, detail='skipping ci' ) # clone repository logging.info('cloning repository: {repo_url}') local_repo_dir = clone_repository(repo_url) logging.info(f'local_repo_dir: {local_repo_dir}') # 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 commands runner_script = load_runner_script(path=runner_script_path) # extract branch name for pipeline_name, pipeline_step in runner_script.pipelines.items(): print(pipeline_name) print(pipeline_step) # send response raise HTTPException( status_code=202, detail='received task' ) 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 )