Auhentication-FastAPI-MemoryDB

!pip install python-jose passlib[bcrypt]

# Build AuthServer
from datetime import datetime, timedelta
from jose import JWTError, jwt


class AuthServer:
    def __init__(self):
        # You can have more flexibilites by putting these info in contructor
        self._access_token_expire_mins = 30
        self._secret_key = "supersecretkey123"
        self._algorithm = "HS256"

    def authen_user(self, *args, **kwargs) -> dict:
        raise NotImplementedError()

    def issue_token(self, *args, **kwargs):
        user_dict = self.authen_user(*args, **kwargs)
        if user_dict is None:
            raise Exception("User not authenticated")

        to_encode = user_dict.copy()
        expire = datetime.now() + timedelta(minutes=self._access_token_expire_mins)
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, self._secret_key, algorithm=self._algorithm)

    def verify_token(self, token: str) -> dict:
        return jwt.decode(token, self._secret_key, algorithms=[self._algorithm])

class FakeAuthServer(AuthServer):
    def authen_user(self, *args, **kwargs):
        return kwargs


# Sample test
auth_server = FakeAuthServer()

token = auth_server.issue_token(user_id="HoHai", password="123")

print(f"Token: {token}")
print(f"Decode token: {auth_server.verify_token(token)}")

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiSG9IYWkiLCJwYXNzd29yZCI6IjEyMyIsImV4cCI6MTc2NDk1MTU0OX0.yikRpE5IZnOa47Ndx5EvMSL_LOz9RQRJG9RZc3w96rU
Decode token: {'user_id': 'HoHai', 'password': '123', 'exp': 1764951549}

import inspect
import asyncio

import nest_asyncio

class AuthUserPassServer(AuthServer):
    def __init__(self, fn_check_userpass):
        super().__init__()
        assert fn_check_userpass is not None, "Must set `fn_check_userpass`"
        self.fn = fn_check_userpass

    # Overriding sync with async breaks Liskov Substitution
    def authen_user(self, username, password) -> dict:
        # Support async functions
        if inspect.iscoroutinefunction(self.fn):

            # Support jupyter
            nest_asyncio.apply()

            loop = asyncio.get_event_loop()
            task = loop.create_task(self.fn(username, password))
            loop.run_until_complete(task)
            return task.result()
        else:
            return self.fn(username, password)

# Sample test
auth_server = AuthUserPassServer(fn_check_userpass=lambda u, p: {"user": u, "password": p})
token = auth_server.issue_token(username="hohai", password="13")

print(f"Token: {token}")
print(f"Decode token: {auth_server.verify_token(token)}")

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoiaG9oYWkiLCJwYXNzd29yZCI6IjEzIiwiZXhwIjoxNzY0OTUyNDMxfQ.48-_p5AEYvbcQYBtPmSdmHzw1QnE9P3DuXbcQqMZM8I
Decode token: {'user': 'hohai', 'password': '13', 'exp': 1764952431}

import asyncio
import inspect

# import nest_asyncio
# nest_asyncio.apply()  # only needed in Jupyter

async def fn_check_user_pass_async(u, p):
    await asyncio.sleep(1.0)
    return {"user": u, "password": p}


auth_server = AuthUserPassServer(fn_check_user_pass_async)

token = auth_server.issue_token(username="hohai", password="13")

print(f"Token: {token}")
print(f"Decode token: {auth_server.verify_token(token)}")


# asyncio.get_event_loop().run_until_complete(coro)


# def wrap_fn():
#     if inspect.iscoroutinefunction(fn_check_user_pass_async):
#         loop = asyncio.get_event_loop()
#         task = loop.create_task(fn_check_user_pass_async("HaiHT", "123"))
#         loop.run_until_complete(task)

#         print(task.result())

#     else:
#         print(fn_check_user_pass_async("HaiHT", "123"))

# wrap_fn()

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoiaG9oYWkiLCJwYXNzd29yZCI6IjEzIiwiZXhwIjoxNzY0OTUyNDU5fQ.ExhPoaR58QpTta9OcVWDlXMt0fWrxUJ8LGUx-2Bm-m4
Decode token: {'user': 'hohai', 'password': '13', 'exp': 1764952459}

import uvicorn
from fastapi import FastAPI
import nest_asyncio
import threading

from fastapi import Depends
from fastapi.security import OAuth2PasswordRequestForm



nest_asyncio.apply()   # Allows nested event loops (required in notebooks)

app = FastAPI()

# Build auth_server: fake for simple, you can change `fn_check_userpass` to refer the crud function
auth_server = AuthUserPassServer(fn_check_userpass=lambda u, p: {"user": u, "password": p})


@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# async def login(form_data):
    token = auth_server.issue_token(form_data.username, form_data.password)
    return {"access_token": token, "token_type": "bearer"}

@app.post("/me")
async def verify(token: str):
    return auth_server.verify_token(token)

def run_app():
    uvicorn.run(app, host="0.0.0.0", port=8000)

thread = threading.Thread(target=run_app, daemon=True, name="Thread run FASTAPI")
thread.start()