You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
148 lines
3.8 KiB
Python
148 lines
3.8 KiB
Python
1 year ago
|
"""Python FastAPI Auth0 integration example
|
||
|
"""
|
||
|
|
||
|
from datetime import datetime
|
||
|
from pydantic import BaseModel
|
||
|
|
||
|
from fastapi import FastAPI, Security
|
||
|
from fastapi.staticfiles import StaticFiles
|
||
|
|
||
|
from .utils import VerifyToken
|
||
|
from .database import TodoDatabase
|
||
|
from .thermal_print import ThermalPrinter
|
||
|
|
||
|
# from wsgiref import simple_server
|
||
|
|
||
|
# Creates app instance
|
||
|
app = FastAPI()
|
||
|
auth = VerifyToken()
|
||
|
data = TodoDatabase()
|
||
|
# printer = ThermalPrinter()
|
||
|
|
||
|
|
||
|
|
||
|
class Todo(BaseModel):
|
||
|
time: str
|
||
|
task: str
|
||
|
recurring: bool
|
||
|
|
||
|
class TodoList(BaseModel):
|
||
|
date: str #Always current date OLD TEXT: #Either current date in %Y-%m-%d format or "recurring" for a recurring task
|
||
|
todos: list[Todo]
|
||
|
|
||
|
|
||
|
class TodoDate(BaseModel):
|
||
|
date: str
|
||
|
# date: str = datetime.today().strftime('%Y-%m-%d')
|
||
|
|
||
|
|
||
|
# @app.get("/api/public")
|
||
|
# def public():
|
||
|
# """No access token required to access this route"""
|
||
|
|
||
|
# result = {
|
||
|
# "status": "success",
|
||
|
# "msg": ("Hello from a public endpoint! You don't need to be "
|
||
|
# "authenticated to see this.")
|
||
|
# }
|
||
|
# return result
|
||
|
|
||
|
|
||
|
@app.get("/api/todos/get")
|
||
|
def get_todos(date: str = datetime.today().strftime('%Y-%m-%d'), auth_result: str = Security(auth.verify)):
|
||
|
"""A valid access token is required to access this route"""
|
||
|
|
||
|
return {
|
||
|
"todos": data.read_todos(date)
|
||
|
}
|
||
|
|
||
|
# result = {
|
||
|
# "todos": [
|
||
|
# {"time": "1:00pm to 2:00pm", "text": "Read a book"},
|
||
|
# {"time": "2:00pm to 3:00pm", "text": "Go for a walk"}
|
||
|
# ]
|
||
|
# } #TODO: replace with database access
|
||
|
|
||
|
# return result
|
||
|
|
||
|
|
||
|
# return auth_result
|
||
|
|
||
|
|
||
|
|
||
|
@app.post("/api/todos/write")
|
||
|
def write_todos(todos: TodoList, auth_result: str = Security(auth.verify)):
|
||
|
"""A valid access token is required to access this route"""
|
||
|
|
||
|
# result = {
|
||
|
# "todos": [
|
||
|
# {"time": "1:00pm to 2:00pm", "text": "Read a book"},
|
||
|
# {"time": "2:00pm to 3:00pm", "text": "Go for a walk"}
|
||
|
# ]
|
||
|
# } #TODO: replace with database write
|
||
|
|
||
|
# return result
|
||
|
|
||
|
# data.save_todo()
|
||
|
|
||
|
print(todos)
|
||
|
|
||
|
# if todos.date != "recurring":
|
||
|
# data.remove_todos(todos.date)
|
||
|
data.remove_todos(todos.date)
|
||
|
data.save_todos(todos.date, todos.todos)
|
||
|
|
||
|
return {"result": todos.date}
|
||
|
|
||
|
# return todos
|
||
|
|
||
|
|
||
|
# @app.post("/api/todos/write_recurring")
|
||
|
# def write_todos_recurring(todos: TodoList, auth_result: str = Security(auth.verify)):
|
||
|
|
||
|
# data.save_todos("recurring", todos.todos)
|
||
|
|
||
|
# return {"result": "success"}
|
||
|
|
||
|
|
||
|
@app.get("/api/todos/print")
|
||
|
def print_todos(date: str = datetime.today().strftime('%Y-%m-%d'), auth_result: str = Security(auth.verify)):
|
||
|
"""A valid access token is required to access this route"""
|
||
|
|
||
|
# result = {
|
||
|
# "todos": [
|
||
|
# {"time": "1:00pm to 2:00pm", "text": "Read a book"},
|
||
|
# {"time": "2:00pm to 3:00pm", "text": "Go for a walk"}
|
||
|
# ]
|
||
|
# } #TODO: replace with access to database if no todos provided from request, otherwise print request data
|
||
|
|
||
|
todos = data.get_todos(date)
|
||
|
printer.print_todos()
|
||
|
|
||
|
return {"result": "success"}
|
||
|
|
||
|
# class SaveTodos:
|
||
|
# def on_post(self, req, resp):
|
||
|
# pass #Handle web interface sending updated todo list
|
||
|
|
||
|
# class PrintTodos:
|
||
|
# def on_post(self, req, resp):
|
||
|
# pass #Handle web interface requesting a print action
|
||
|
|
||
|
|
||
|
# @app.get("/api/get_todos-scoped")
|
||
|
# def private_scoped(auth_result: str = Security(auth.verify, scopes=['todos:all'])):
|
||
|
# """A valid access token and an appropriate scope are required to access
|
||
|
# this route
|
||
|
# """
|
||
|
|
||
|
# return auth_result
|
||
|
|
||
|
|
||
|
app.mount("/", StaticFiles(directory="application/static",html = True), name="static")
|
||
|
|
||
|
|
||
|
# if __name__ == '__main__':
|
||
|
# print('ay')
|
||
|
# httpd = simple_server.make_server('127.0.0.1', 8000, app)
|
||
|
# httpd.serve_forever()
|