106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
from datetime import timedelta
|
|
|
|
import uvicorn
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from starlette import status
|
|
from starlette.middleware.cors import CORSMiddleware
|
|
|
|
from constants import CLIENT_OPERATION_QUIT, EXIT_COMMANDS, SHEERKA_PORT
|
|
from core.Sheerka import Sheerka
|
|
from server.authentication import ACCESS_TOKEN_EXPIRE_MINUTES, User, authenticate_user, create_access_token, \
|
|
fake_users_db, get_current_active_user
|
|
|
|
app = FastAPI()
|
|
origins = [
|
|
"http://localhost:56426",
|
|
"http://localhost:5173",
|
|
]
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# NEED TO FIND A WAY TO INIT SHEERKA within the __name__ == "__main__" section
|
|
# As of now, if we do that, sheerka is not properly initialized using the command
|
|
# 'uvicorn server.main:app' from the shell
|
|
sheerka = Sheerka()
|
|
sheerka.initialize()
|
|
|
|
|
|
@app.get("/")
|
|
async def root() -> str:
|
|
"""
|
|
Root path. Simply display a welcome message
|
|
:return:
|
|
:rtype:
|
|
"""
|
|
return "Welcome, my name is Sheerka."
|
|
|
|
|
|
@app.post("/token")
|
|
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
|
|
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires)
|
|
return {"first_name": user.firstname,
|
|
"last_name": user.lastname,
|
|
"access_token": access_token,
|
|
"token_type": "bearer"}
|
|
|
|
|
|
@app.post("/echo/{message}", status_code=status.HTTP_200_OK, response_model=dict)
|
|
async def echo(message: str, current_user: User = Depends(get_current_active_user)) -> dict:
|
|
"""
|
|
|
|
:param current_user:
|
|
:type current_user:
|
|
:param message:
|
|
:type message:
|
|
:return:
|
|
:rtype:
|
|
"""
|
|
if message in EXIT_COMMANDS:
|
|
return {
|
|
"status": True,
|
|
"response": "Take care.",
|
|
"command": CLIENT_OPERATION_QUIT
|
|
}
|
|
|
|
return {
|
|
"status": True,
|
|
"response": f"from {current_user.username}: {message}",
|
|
"command": None,
|
|
}
|
|
|
|
|
|
@app.post("/command/{message}", status_code=status.HTTP_200_OK, response_model=dict)
|
|
async def command(message: str, current_user: User = Depends(get_current_active_user)) -> dict:
|
|
if message in EXIT_COMMANDS:
|
|
return {
|
|
"status": True,
|
|
"response": "Take care.",
|
|
"command": CLIENT_OPERATION_QUIT
|
|
}
|
|
res = sheerka.evaluate_user_input(message, current_user)
|
|
return {
|
|
"status": res[0].status,
|
|
"response": res[0].value.body,
|
|
"command": None
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run("server.main:app", port=SHEERKA_PORT, log_level="info")
|