del fastapi-users

server/main-server/src/main.py

@@ -1,13 +1,13 @@
 import asyncio
-from fastapi import Depends, FastAPI, HTTPException, Request
-from fastapi.responses import RedirectResponse
-from database.db import Database
-import uvicorn
-from src.database.schemas import *
-from src.auth.transport import Transport
-from src.database.user import User
+from contextlib import asynccontextmanager
 
-app = FastAPI(title='sclient-main-server')
+from authx import AuthXConfig, AuthX, RequestToken
+from fastapi import Response, FastAPI, Request, HTTPException, Depends
+from fastapi.responses import RedirectResponse
+import uvicorn
+
+from src.database.db import Database
+from src.database.user import User, UserLogin
 
 ### Settings
 # TODO: Create .env
@@ -18,74 +18,80 @@ ADMIN_PASSWORD = 'admin'
 DATABASE_USER = 'ADMIN'
 DATABASE_PASS = '123123'
 DATABASE_HOST = '127.0.0.1'
-DATABASE_PORT = 5432
 DATABASE_NAME = 'sonoma-db'
 SECRET = 'SECRET'
 
 ###
 
-db = Database(
-        DATABASE_USER,
-        DATABASE_PASS,
-        DATABASE_HOST,
-        DATABASE_PORT,
-        DATABASE_NAME,
-        SECRET
-    )
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await db.connect()
+    await db.init()
+    yield
+    await db.disconnect()
+
+db = Database(
+    ADMIN_NAME,
+    ADMIN_PASSWORD,
+    DATABASE_USER,
+    DATABASE_PASS,
+    DATABASE_HOST,
+    DATABASE_NAME
+)
+
+app = FastAPI(title='sclient-main-server', lifespan=lifespan)
+
+config = AuthXConfig()
+config.JWT_SECRET_KEY = SECRET
+config.JWT_ACCESS_COOKIE_NAME = "sclient_access_token"
+config.JWT_TOKEN_LOCATION = ["cookies"]
+
+security = AuthX(config=config)
+security.handle_errors(app)
 
-transport = Transport(SECRET, db)
 
 class App:
-    def init(self, loop) -> None:
-        config = uvicorn.Config(
-            app,
-            loop=loop,
-            host='0.0.0.0',
-            port=PORT
-        )
-        server = uvicorn.Server(config)
-        loop.run_until_complete(server.serve())
 
     @app.get('/')
     async def docs(self: Request):
         return RedirectResponse(f'{self.url}docs')
 
-    @app.get("/authenticated-route")
-    async def authenticated_route(user: User = Depends(transport.current_active_user)):
-        return {"message": f"Hello {user.email}!"}
+    ### Auth
+
+    @app.post('/login')
+    async def login(self: Request, credentials: UserLogin, response: Response):
+        user = await db.get_user(credentials.username)
+
+        if user is not None:
+            if user[1] == credentials.password:
+                token = security.create_access_token(uid=credentials.username)
+                response.set_cookie(config.JWT_ACCESS_COOKIE_NAME, token)
+                return {
+                    "access_token": token
+                }
+        raise HTTPException(
+            401,
+            detail='Incorrect username or password'
+        )
+    ###
+
+    ### Protected
+
+    @app.get('/protected/auth', dependencies=[Depends(security.access_token_required)])
+    async def auth(self: Request):
+        try:
+            return {"message": "Hello world !"}
+        except Exception as e:
+            raise HTTPException(
+                401,
+                detail={"message": str(e)}
+            ) from e
+
+    ###
+
 
 def main():
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-
-    server = App()
-
-    app.include_router(
-        transport.fastapi_users.get_auth_router(transport.auth_backend), prefix="/auth/jwt", tags=["auth"]
-    )
-    app.include_router(
-        transport.fastapi_users.get_register_router(UserRead, UserCreate),
-        prefix="/auth",
-        tags=["auth"],
-    )
-    app.include_router(
-        transport.fastapi_users.get_reset_password_router(),
-        prefix="/auth",
-        tags=["auth"],
-    )
-    app.include_router(
-        transport.fastapi_users.get_verify_router(UserRead),
-        prefix="/auth",
-        tags=["auth"],
-    )
-    app.include_router(
-        transport.fastapi_users.get_users_router(UserRead, UserUpdate),
-        prefix="/users",
-        tags=["users"],
-    )
-
-    loop.run_until_complete(db.create_db_and_tables())
-    server.init(loop)
+    uvicorn.run(app, host='0.0.0.0', port=PORT)
 
 if __name__ == '__main__':
     main()