53 lines
1.3 KiB
Python
53 lines
1.3 KiB
Python
# from waitress import serve
|
|
# from flask import Flask, request, jsonify
|
|
import uvicorn
|
|
from fastapi import FastAPI, Request
|
|
from dotenv import load_dotenv
|
|
import logging
|
|
from pydantic import BaseModel
|
|
from typing import List
|
|
|
|
from utils import clone_repository
|
|
|
|
class GiteaRequest(BaseModel):
|
|
ref: str
|
|
before: str
|
|
after: str
|
|
compare_url: str
|
|
commits: List[dict]
|
|
total_commits: int
|
|
head_commit: dict
|
|
repository: dict
|
|
pusher: dict
|
|
sender: dict
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get('/')
|
|
async def root():
|
|
return '<h1>Welcome to Gitea Runner</h1>'
|
|
|
|
@app.post('/test')
|
|
async def test(request: Request):
|
|
""" Test connection to server. """
|
|
logging.debug('Content-Type: ', request.headers.get('content-type'))
|
|
return {'status': 'success'}
|
|
|
|
@app.post('/event')
|
|
def handle_event(request: GiteaRequest):
|
|
""" Handle event from webhook. """
|
|
# stop early if allowed
|
|
commit_msg = request.head_commit['message']
|
|
if commit_msg.lower().startswith('[skip ci]'):
|
|
return {'status': 'received task'}
|
|
# clone repository
|
|
repo_url = request.repository['clone_url']
|
|
local_repo_dir = clone_repository(repo_url)
|
|
print('local_repo_dir: ', local_repo_dir)
|
|
return {'status': 'success'}
|
|
|
|
if __name__ == '__main__':
|
|
uvicorn.run(
|
|
app=app,
|
|
host='0.0.0.0'
|
|
) |