79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
import uuid
|
|
|
|
from sqlmodel import select
|
|
from fastapi import Depends
|
|
from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin
|
|
from fastapi_users.authentication import BearerTransport, AuthenticationBackend
|
|
from fastapi_users.authentication.strategy.db import AccessTokenDatabase, DatabaseStrategy
|
|
|
|
from user.models import User, get_user_db, AccessToken, get_access_token_db
|
|
|
|
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):
|
|
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}""")
|