import uvicorn from fastapi import FastAPI, Request, HTTPException, BackgroundTasks from fastapi.responses import HTMLResponse from dotenv import load_dotenv import logging import os import httpx 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 = """ Gitea Runner

Welcome to Gitea Runner

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('/') def handle_event(request: GiteaRequest, background_tasks: BackgroundTasks): """ 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 / # '.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 ) # 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, workers=1 )