feat: add basic authentications with fastapi-users

fastapi/.gitignore

@@ -157,4 +157,6 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/
+
+*.env

fastapi/app/auth/__init__.py

@@ -0,0 +1,3 @@
+from .routes import routers as AuthRouters
+
+__all__ = ['AuthRouters']

fastapi/app/auth/libs.py

@@ -0,0 +1,57 @@
+from typing import Optional
+
+from app.configs import Configs
+from beanie import PydanticObjectId
+from fastapi import Depends, Request
+from fastapi_users import BaseUserManager, FastAPIUsers
+from fastapi_users.authentication import (AuthenticationBackend,
+                                          BearerTransport, JWTStrategy)
+from fastapi_users.db import BeanieUserDatabase, ObjectIDIDMixin
+
+from .models import User
+
+
+async def get_user_db():
+    yield BeanieUserDatabase(User)
+
+
+class UserManager(ObjectIDIDMixin, BaseUserManager[User, PydanticObjectId]):
+    reset_password_token_secret = Configs.SECRET_KEY
+    verification_token_secret = Configs.SECRET_KEY
+
+    async def on_after_register(self, user: User, request: Optional[Request] = None):
+        print(f"User {user.id} has registered.")
+
+    async def on_after_forgot_password(
+        self, user: User, token: str, request: Optional[Request] = None
+    ):
+        print(f"User {user.id} has forgot their password. Reset token: {token}")
+
+    async def on_after_request_verify(
+        self, user: User, token: str, request: Optional[Request] = None
+    ):
+        print(
+            f"Verification requested for user {user.id}. Verification token: {token}")
+
+
+async def get_user_manager(user_db: BeanieUserDatabase = Depends(get_user_db)):
+    yield UserManager(user_db)
+
+
+def get_jwt_strategy() -> JWTStrategy:
+    return JWTStrategy(secret=Configs.SECRET_KEY,
+                       lifetime_seconds=Configs.ACCESS_TOKEN_EXPIRE_MINUTES * 60)
+
+
+bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")
+
+auth_backend = AuthenticationBackend(
+    name="jwt",
+    transport=bearer_transport,
+    get_strategy=get_jwt_strategy,
+)
+
+fastapi_users = FastAPIUsers[User, PydanticObjectId](
+    get_user_manager, [auth_backend])
+
+current_active_user = fastapi_users.current_user(active=True)

fastapi/app/auth/models.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+from enum import Enum
+from typing import List, Optional
+
+from beanie import PydanticObjectId
+from pydantic import Field, EmailStr
+from app.base.models import AppBaseModel
+from fastapi_users import schemas
+from fastapi_users.db import BeanieBaseUser, BaseOAuthAccount
+
+
+class SocialScope(str, Enum):
+    email: str = "email"
+    google: str = "google"
+
+
+class UserRead(schemas.BaseUser[PydanticObjectId]):
+    pass
+
+
+class UserCreate(schemas.BaseUserCreate):
+    pass
+
+
+class UserUpdate(schemas.BaseUserUpdate):
+    pass
+
+
+class User(BeanieBaseUser[PydanticObjectId], AppBaseModel):
+    email: EmailStr
+    username: Optional[str] = Field(None, description='Username')
+    first_name: Optional[str] = Field(None)
+    last_name: Optional[str] = Field(None)
+    picture: Optional[str] = Field(None)
+
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+    last_login_at: datetime = Field(default_factory=datetime.now)

fastapi/app/auth/routes.py

@@ -0,0 +1,29 @@
+from fastapi import APIRouter, Depends
+from .models import User, UserCreate, UserRead, UserUpdate
+from .libs import auth_backend, current_active_user, fastapi_users
+
+CLIENT_REDIRECT_URL = "http://localhost:3000/auth/google"
+GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
+GOOGLE_TOKEN_API = "https://oauth2.googleapis.com/token"
+
+router = APIRouter()
+
+get_auth_router = fastapi_users.get_auth_router(auth_backend)
+get_register_router = fastapi_users.get_register_router(UserRead, UserCreate)
+get_reset_password_router = fastapi_users.get_reset_password_router()
+get_verify_router = fastapi_users.get_verify_router(UserRead)
+get_users_router = fastapi_users.get_users_router(UserRead, UserUpdate)
+
+routers = [
+    (router, dict(prefix="/auth", tags=["auth"])),
+    (get_auth_router, dict(prefix="/auth/jwt", tags=["auth"])),
+    (get_register_router, dict(prefix="/auth", tags=["auth"])),
+    (get_reset_password_router, dict(prefix="/auth", tags=["auth"])),
+    (get_verify_router, dict(prefix="/auth", tags=["auth"])),
+    (get_users_router, dict(prefix="/users", tags=["users"])),
+]
+
+
+@router.get("/authenticated-route")
+async def authenticated_route(user: User = Depends(current_active_user)):
+    return {"message": f"Hello {user.email}!"}

fastapi/app/auth/schemas.py

@@ -0,0 +1,14 @@
+from beanie import PydanticObjectId
+from fastapi_users import schemas
+
+
+class UserRead(schemas.BaseUser[PydanticObjectId]):
+    pass
+
+
+class UserCreate(schemas.BaseUserCreate):
+    pass
+
+
+class UserUpdate(schemas.BaseUserUpdate):
+    pass

fastapi/app/base/models.py

@@ -0,0 +1,31 @@
+import json
+from datetime import datetime
+
+from bson import ObjectId
+from pydantic import BaseModel
+
+
+class PyObjectId(ObjectId):
+    @classmethod
+    def __get_validators__(cls):
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, v):
+        if not ObjectId.is_valid(v):
+            raise ValueError("Invalid objectid")
+        return ObjectId(v)
+
+    @classmethod
+    def __modify_schema__(cls, field_schema):
+        field_schema.update(type="string")
+
+
+class AppBaseModel(BaseModel):
+    class Config:
+        json_encoders = {
+            datetime: lambda dt: dt.isoformat()
+        }
+
+    def json(self):
+        return json.loads(json.dumps(self.dict(), default=str))

fastapi/app/configs.py

@@ -0,0 +1,46 @@
+import os
+from functools import lru_cache
+from typing import List
+
+from pydantic import BaseSettings, Field
+
+
+def get_env_file():
+    stage = os.environ.get('ENV') or 'dev'
+    return f'{stage}.env'
+
+
+class Settings(BaseSettings):
+    DEBUG: bool = False
+
+    APP_NAME: str = "The Endings"
+    HTTPS: bool = False
+    HOST: str = "localhost"
+
+    SECRET_KEY: str
+    ALGORITHM: str
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+
+    DB_DATABASE: str
+    DB_URL: str
+
+    ORIGINS: List[str] = Field(['http://localhost'], env='ORIGINS')
+    ALLOWED_HOSTS: List[str] = Field(..., env='ALLOWED_HOSTS')
+
+    class Config:
+        env_file = get_env_file()
+
+    @property
+    def URL(self) -> str:
+        protocol = 'https' if self.HTTPS else 'http'
+        return f'{protocol}://{self.HOST}'
+
+
+Configs = Settings()
+
+print('Configs:\n', Configs)
+
+
+@lru_cache()
+def get_settings():
+    return Configs

fastapi/app/db.py

@@ -0,0 +1,19 @@
+from beanie import init_beanie
+from motor import motor_asyncio
+
+from .auth.models import User
+from .configs import Configs
+
+client = motor_asyncio.AsyncIOMotorClient(
+    Configs.DB_URL, uuidRepresentation="standard"
+)
+database = client[Configs.DB_DATABASE]
+
+
+async def on_startup():
+    await init_beanie(
+        database=database,
+        document_models=[
+            User,
+        ],
+    )

fastapi/index.md

@@ -40,3 +40,16 @@ python -m uvicorn main:app --reload
 Interactive Docs (Swagger UI) - http://127.0.0.1:8000/docs
 
 테스트할 때는 직접 API call도 할 수 있는 interactive docs가 좋다고 생각한다.
+
+
+# FastAPI-users 설치
+
+```
+pip install fastapi-users[beanie]
+```
+
+# env 파일 읽도록 설정
+
+```
+pip install pydantic[dotenv]
+```

fastapi/main.py

@@ -1,3 +1,14 @@
 from fastapi import FastAPI
 
+from app import db
+from app.auth import AuthRouters
+
 app = FastAPI()
+
+for router, kwargs in AuthRouters:
+    app.include_router(router=router, **kwargs)
+
+
+@app.on_event("startup")
+async def on_startup():
+    await db.on_startup()