82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
|
||
import asyncio
|
||
#from sqlalchemy import create_engine #Не async
|
||
from sqlalchemy.orm import DeclarativeBase, sessionmaker
|
||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
||
from sqlalchemy import Column, Integer, String, Boolean, select
|
||
|
||
from pathlib import Path
|
||
db_folder = Path(__file__).parent / "DB"
|
||
db_folder.mkdir(parents=True, exist_ok=True)
|
||
db_path = db_folder / "example.db"
|
||
async_engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}", echo=True)
|
||
#sqlite+aiosqlite — тип БД + async-драйвер ///example.db — путь к файлу (три слэша, если путь относительный; четыре, если абсолютный
|
||
|
||
from passlib.context import CryptContext
|
||
#Hash password
|
||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
def hash_password(password: str) -> str:
|
||
return pwd_context.hash(password)
|
||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||
return pwd_context.verify(plain_password, hashed_password)
|
||
|
||
class Base(DeclarativeBase):
|
||
pass
|
||
|
||
AsyncSessionLocal = sessionmaker(async_engine,class_=AsyncSession, expire_on_commit=False)
|
||
|
||
class User(Base):
|
||
__tablename__ = "users"
|
||
|
||
id = Column(Integer, primary_key=True, index=True)
|
||
email = Column(String(254), unique=True, nullable=False)
|
||
description = Column(String, nullable=False)
|
||
activated = Column(Boolean, default=False)
|
||
password = Column(String, nullable=False)
|
||
|
||
async def init_db():
|
||
async with async_engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
async def CreateUser(user_info):
|
||
async with AsyncSessionLocal() as session:
|
||
new_user = User(email=user_info.email, description=user_info.description, activated=user_info.activated, password=hash_password(user_info.password))
|
||
session.add(new_user)
|
||
await session.commit()
|
||
await session.refresh(new_user)
|
||
print(new_user.id)
|
||
async def GetUser(id):
|
||
async with AsyncSessionLocal() as session:
|
||
result = await session.execute(select(User).where(User.id==id))
|
||
user = result.scalar_one_or_none()
|
||
return user
|
||
async def get_all_rows():
|
||
async with AsyncSessionLocal() as session:
|
||
result = await session.execute(select(User))
|
||
users = result.scalars().all()
|
||
return users
|
||
async def UpdateUser(user_info):
|
||
async with AsyncSessionLocal() as session:
|
||
result = await session.execute(select(User).where(User.id==user_info.id))
|
||
user = result.scalar_one_or_none()
|
||
if user:
|
||
user.email = user_info.email
|
||
user.description = user_info.description
|
||
user.activated = user_info.activated
|
||
user.password = hash_password(user_info.password)
|
||
await session.commit()
|
||
async def DeleteUser(id):
|
||
async with AsyncSessionLocal() as session:
|
||
result = await session.execute(select(User).where(User.id==id))
|
||
user = result.scalar_one_or_none()
|
||
if user:
|
||
await session.delete(user)
|
||
await session.commit()
|
||
async def main():
|
||
await init_db()
|
||
await CreateUser()
|
||
await get_all_rows()
|
||
# await UpdateUser(1)
|
||
# await GetUser(1)
|
||
# await DeleteUser(1)
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |