Initial commit

api/app/user/manager.py

@@ -0,0 +1,80 @@
+import uuid
+
+from sqlmodel import select
+from fastapi import Depends
+from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin, models, exceptions, schemas
+from fastapi_users.authentication import BearerTransport, AuthenticationBackend
+from fastapi_users.authentication.strategy.db import AccessTokenDatabase, DatabaseStrategy
+
+from .models import User, get_user_db, AccessToken, get_access_token_db, UserRead, UserUpdate, UserCreate
+from db import get_session
+
+SECRET = "SECRET"
+TOKEN_LIFETIME = 3600
+
+bearer_transport = BearerTransport(tokenUrl="auth/login")
+
+class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
+    pass
+
+
+async def get_user_manager(user_db=Depends(get_user_db)) -> UserManager:
+    yield UserManager(user_db)
+
+def get_database_strategy(
+    access_token_db: AccessTokenDatabase[AccessToken] = Depends(get_access_token_db),
+) -> DatabaseStrategy:
+    return DatabaseStrategy(access_token_db, lifetime_seconds=TOKEN_LIFETIME)
+
+auth_backend = AuthenticationBackend(
+    name="db",
+    transport=bearer_transport,
+    get_strategy=get_database_strategy,
+)
+
+fastapi_users = FastAPIUsers[User, uuid.UUID](
+    get_user_manager,
+    [auth_backend],
+)
+
+
+get_current_user = fastapi_users.current_user(active=True)
+get_current_superuser = fastapi_users.current_user(active=True, superuser=True)
+
+
+#user_router = fastapi_users.get_users_router(UserRead, UserUpdate)
+#user_router.include_router(fastapi_users.get_reset_password_router())
+
+reset_password_router = fastapi_users.get_reset_password_router()
+auth_router = fastapi_users.get_auth_router(auth_backend)
+
+
+def create_admin_account():
+    session = get_session().__next__()
+    admin_email = 'root@root.fr'
+    statement = select(User).where(User.email == admin_email).limit(1)
+    admin_user = session.exec(statement).first()
+
+    if admin_user is not None:
+        return
+
+    import secrets
+
+    from fastapi_users.password import PasswordHelper
+
+    password_length = 16
+    password = secrets.token_urlsafe(password_length)
+
+    admin_user = User(
+        id=uuid.uuid4(),
+        email=admin_email,
+        hashed_password=PasswordHelper().hash(password),
+        is_active=True,
+        is_superuser=True,
+        is_verified=True
+    )
+    session.add(admin_user)
+    session.commit()
+    print(f"""Admin account created:
+login: {admin_email}
+password: {password}""")