feat: authentication — register, login, JWT, tests

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-03 22:08:28 +05:00
parent 5b92553148
commit bfb1fc9312
11 changed files with 300 additions and 0 deletions
View File
+52
View File
@@ -0,0 +1,52 @@
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.user import User
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_access_token(user_id: str) -> str:
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
payload = {"sub": user_id, "exp": expire}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def decode_access_token(token: str) -> str | None:
try:
payload = jwt.decode(
token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
)
return payload.get("sub")
except JWTError:
return None
async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
result = await db.execute(select(User).where(User.username == username))
return result.scalar_one_or_none()
async def create_user(db: AsyncSession, username: str, email: str, password: str) -> User:
user = User(
username=username,
email=email,
password_hash=hash_password(password),
)
db.add(user)
await db.commit()
await db.refresh(user)
return user