47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import logging
|
|
import structlog
|
|
from unicodedata import name
|
|
from urllib.error import HTTPError
|
|
from fastapi import Depends, FastAPI, Request, BackgroundTasks
|
|
from pydantic import BaseModel
|
|
from app.authentik.api import Authentik
|
|
from app.authentik.models import User
|
|
from app.event_controller import Authentik_Hook_Model, Event_Controller, Http_request
|
|
from .wekan.api import Wekan
|
|
import json
|
|
|
|
logging = structlog.get_logger()
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {'message': 'Hello World'}
|
|
|
|
@app.post("/authentik/hook/")
|
|
async def hook(model: Authentik_Hook_Model,
|
|
http_request: Http_request,
|
|
event_controller: Event_Controller = Depends()):
|
|
logging.info(model)
|
|
logging.info(http_request)
|
|
if http_request.path == "/api/v3/core/users/":
|
|
event_controller.handle_model_created_event(model)
|
|
return 200
|
|
|
|
# @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
|