数据库模型基本完成

This commit is contained in:
carry 2025-01-19 23:45:13 +08:00
parent fb41b442da
commit 287a1e5629
5 changed files with 74 additions and 0 deletions

5
models/__init__.py Normal file
View File

@ -0,0 +1,5 @@
from .user import User
from .auth import get_password_hash, verify_password
from .database import init_db
__all__ = ["User", "get_password_hash", "verify_password", "init_db"]

11
models/auth.py Normal file
View File

@ -0,0 +1,11 @@
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password: str) -> str:
"""Generate password hash using bcrypt"""
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify password against stored hash"""
return pwd_context.verify(plain_password, hashed_password)

25
models/database.py Normal file
View File

@ -0,0 +1,25 @@
from sqlalchemy.ext.asyncio import AsyncEngine
from .user import Base, User, UserRole
from sqlalchemy import select
from .auth import get_password_hash
async def init_db(engine: AsyncEngine):
"""Initialize database"""
async with engine.begin() as conn:
# Create all tables
await conn.run_sync(Base.metadata.create_all)
# Check if system admin exists
result = await conn.execute(
select(User).where(User.role == UserRole.SYSTEM_ADMIN)
)
if not result.scalars().first():
# Create default system admin
admin = User(
username="admin",
password=get_password_hash("password"),
role=UserRole.SYSTEM_ADMIN,
description="default system admin"
)
conn.add(admin)
await conn.commit()

25
models/user.py Normal file
View File

@ -0,0 +1,25 @@
from datetime import datetime, timezone
from enum import Enum
from sqlalchemy import Column, Integer, String, Text, DateTime, Enum as SQLEnum
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class UserRole(str, Enum):
SYSTEM_ADMIN = "system_admin"
ADMIN = "admin"
USER = "user"
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), unique=True, nullable=False)
password = Column(String(255), nullable=False)
role = Column(SQLEnum(UserRole), nullable=False, default=UserRole.USER)
description = Column(Text)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
def __repr__(self):
return f"<User(id={self.id}, username={self.username}, role={self.role})>"

8
requirements.txt Normal file
View File

@ -0,0 +1,8 @@
fastapi>=0.95.2
sqlalchemy>=2.0.15
passlib>=1.7.4
bcrypt>=4.0.1
pyjwt>=2.6.0
pytest>=7.3.1
requests>=2.28.2
uvicorn[standard]>=0.21.1