|
from sqlalchemy import Column, Integer, String |
|
from sqlalchemy.ext.asyncio import AsyncSession |
|
from sqlalchemy.orm import declarative_base |
|
|
|
Base = declarative_base() |
|
|
|
class User(Base): |
|
__tablename__ = "users" |
|
|
|
id = Column(Integer, primary_key=True) |
|
username = Column(String, unique=True) |
|
password = Column(String) |
|
profile = Column(String) |
|
tags = Column(String) |
|
|
|
async def create(self): |
|
async with AsyncSession() as session: |
|
session.add(self) |
|
await session.commit() |
|
return self |
|
|
|
async def read_all(): |
|
async with AsyncSession() as session: |
|
return await session.execute(select(User)).scalars().all() |
|
|
|
async def read_one(user_id: int): |
|
async with AsyncSession() as session: |
|
return await session.get(User, user_id) |
|
|
|
async def update(self, user_id: int, user: User): |
|
async with AsyncSession() as session: |
|
user = await session.get(User, user_id) |
|
user.username = user.username |
|
user.password = user.password |
|
user.profile = user.profile |
|
user.tags = user.tags |
|
await session.commit() |
|
return user |
|
|
|
async def delete(user_id: int): |
|
async with AsyncSession() as session: |
|
user = await session.get(User, user_id) |
|
await session.delete(user) |
|
await session.commit() |