41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import uuid
|
|
|
|
from motor.motor_asyncio import AsyncIOMotorDatabase
|
|
|
|
from core.security import get_password_hash, verify_password
|
|
from crud.base import CRUDBase
|
|
from schemas import UserCreate, UserDBRW
|
|
|
|
__all__ = 'user',
|
|
|
|
|
|
class CRUDUser(CRUDBase):
|
|
|
|
async def get_by_user(self, db: AsyncIOMotorDatabase, name: str):
|
|
res = await db[self.coll_name].find_one({'name': name})
|
|
return res
|
|
|
|
async def create(self, db: AsyncIOMotorDatabase, obj_in: UserCreate):
|
|
db_obj = UserDBRW(
|
|
email=obj_in.email,
|
|
hashed_password=get_password_hash(obj_in.password),
|
|
name=obj_in.name,
|
|
is_superuser=obj_in.is_superuser,
|
|
_id=uuid.uuid1().hex
|
|
)
|
|
return await db[self.coll_name].insert_one(db_obj.dict(by_alias=True))
|
|
|
|
async def authenticate(self, db: AsyncIOMotorDatabase, name: str, password: str):
|
|
user_obj = UserDBRW(**await self.get_by_user(db, name=name))
|
|
if not user_obj:
|
|
return None
|
|
if not verify_password(password, user_obj.hashed_password):
|
|
return None
|
|
return user_obj
|
|
|
|
async def create_index(self, db: AsyncIOMotorDatabase):
|
|
await db[self.coll_name].create_index('username', unique=True)
|
|
|
|
|
|
user = CRUDUser('user')
|