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" ) # execute runner script logging.info('rexecuting runner script') execute_runner_script(path=runner_script_path) # # 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 )