56 lines
1.4 KiB
Python
56 lines
1.4 KiB
Python
import logging
|
|
import structlog
|
|
from unicodedata import name
|
|
from urllib.error import HTTPError
|
|
from fastapi import FastAPI, Request
|
|
from pydantic import BaseModel
|
|
|
|
from app.authentik.api import Authentik
|
|
from app.authentik.models import User
|
|
from .wekan.api import Wekan
|
|
import json
|
|
|
|
logging = structlog.get_logger()
|
|
|
|
class Hook_Model(BaseModel):
|
|
pk: str
|
|
app: str
|
|
name: str
|
|
model_name: str
|
|
|
|
class Http_request(BaseModel):
|
|
args: dict
|
|
path: str
|
|
method: str
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {'message': 'Hello World'}
|
|
|
|
|
|
@app.post("/authentik/hook/")
|
|
async def hook(model: Hook_Model, http_request: Http_request):
|
|
# print(await request.body())
|
|
logging.info(model)
|
|
logging.info(http_request)
|
|
return 200
|
|
# model_created = json.loads(r['body'].split("model_created: ")[1])["model"]
|
|
# hook wekan.create_user(model_created["pk"])
|
|
|
|
@app.get("/authentik/create_hook/")
|
|
async def hook(request: Request):
|
|
a = Authentik(base="http://localhost:9000/", token="foobar123")
|
|
res = a.create_web_hook()
|
|
logging.info(res)
|
|
|
|
@app.get("/authentik/users/create_demo_user/")
|
|
async def create_demo_user(request: Request):
|
|
a = Authentik(base="http://localhost:9000/", token="foobar123")
|
|
try:
|
|
user = a.create_user(User(username="demo", name="dmeo", email="foo@example.org"))
|
|
except Exception as e: # TODO
|
|
return e
|
|
logging.info(user)
|
|
return user.dict |