97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
|
from sqlalchemy.orm import DeclarativeBase, sessionmaker
|
|
from sqlalchemy import Column, Integer, String, Boolean, select,func, DateTime
|
|
import asyncio
|
|
from datetime import datetime,timezone
|
|
|
|
from pathlib import Path
|
|
db_folder = Path(__file__).parent / "DB"
|
|
db_folder.mkdir(parents=True, exist_ok=True)
|
|
db_path = db_folder / "guests.db"
|
|
|
|
async_engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}", echo=True)
|
|
|
|
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)
|
|
code = Column(String, unique=True, nullable=True)
|
|
|
|
name = Column(String, nullable=True)
|
|
surname = Column(String, nullable=True)
|
|
text_field = Column(String, nullable=True)
|
|
food = Column(Boolean)
|
|
alco = Column(Boolean)
|
|
types_of_alco = Column(String, default="Nothing")
|
|
|
|
activated = Column(Boolean)
|
|
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
|
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
|
|
last_login = Column(DateTime(timezone=True))
|
|
|
|
admin = Column(Boolean, default=False)
|
|
|
|
async def create_user(user_info):
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(select(User).where(User.code==user_info.code))
|
|
user = result.scalar_one_or_none()
|
|
if user == None:
|
|
user_data = user_info.dict(exclude_unset=True)
|
|
new_user = User(**user_data)
|
|
session.add(new_user)
|
|
await session.commit()
|
|
await session.refresh(new_user)
|
|
return user
|
|
|
|
async def update_user(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:
|
|
update_data = user_info.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
if hasattr(user, key):
|
|
setattr(user, key, value)
|
|
await session.commit()
|
|
return user
|
|
|
|
async def list_users():
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(select(User))
|
|
users = result.scalars().all()
|
|
if users:
|
|
return users
|
|
else:
|
|
return None
|
|
async def list_user(id):
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(select(User).where(User.id == id))
|
|
user = result.scalar_one_or_none()
|
|
if user:
|
|
return user
|
|
else:
|
|
return None
|
|
async def list_user_by_code(code):
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(select(User).where(User.code == code))
|
|
user = result.scalar_one_or_none()
|
|
if user:
|
|
return user
|
|
else:
|
|
return None
|
|
async def login_user(code):
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(select(User).where(User.code == code.code))
|
|
user = result.scalar_one_or_none()
|
|
if user:
|
|
user.last_login=datetime.now(timezone.utc)
|
|
user.activated=True
|
|
await session.commit()
|
|
return user
|
|
else:
|
|
return None |