Feat 초기 개발분

Subject
Feat	새로운 기능을 추가
Fix	버그 수정
Design	CSS 등 사용자 UI 디자인 변경
!BREAKING CHANGE	커다란 API 변경의 경우
!HOTFIX	급하게 치명적인 버그를 고쳐야하는 경우
Style	코드 포맷 변경, 세미 콜론 누락, 코드 수정이 없는 경우
Refactor	프로덕션 코드 리팩토링
Comment	필요한 주석 추가 및 변경
Docs	문서 수정
Test	테스트 코드, 리펙토링 테스트 코드 추가, Production Code(실제로 사용하는 코드) 변경 없음
Chore	빌드 업무 수정, 패키지 매니저 수정, 패키지 관리자 구성 등 업데이트, Production Code 변경 없음
Rename	파일 혹은 폴더명을 수정하거나 옮기는 작업만인 경우
Remove	파일을 삭제하는 작업만 수행한 경우
---
Footer
Fixes: 이슈 수정중 (아직 해결되지 않은 경우)
Resolves: 이슈를 해결했을 때 사용
Ref: 참고할 이슈가 있을 때 사용
Related to: 해당 커밋에 관련된 이슈번호 (아직 해결되지 않은 경우)

core/database.py

@@ -0,0 +1,21 @@
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+from starlette.config import Config
+
+config = Config('.env')
+SQLALCHEMY_DATABASE_URL = config('SQLALCHEMY_DATABASE_URL')
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

core/models/__init__.py

@@ -0,0 +1 @@
+from core.models import models

core/models/models.py

@@ -0,0 +1,15 @@
+from sqlalchemy import Column, Integer, ForeignKey, String, DateTime
+from sqlalchemy.orm import relationship
+
+from core.database import Base
+
+
+class User(Base):
+    __tablename__ = "user"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    stdId = Column(String, unique=True, nullable=False)
+    password = Column(String, nullable=False)
+    name = Column(String, nullable=False)
+    authority = Column(Integer, nullable=False)
+    hide = Column(Integer, nullable=False)

core/schemas/__init__.py

@@ -0,0 +1,4 @@
+from core.schemas import playlist_info
+from core.schemas import song
+from core.schemas import user
+

core/schemas/user.py

@@ -0,0 +1,53 @@
+from pydantic import BaseModel, field_validator
+
+
+class User(BaseModel):
+    id: int
+    stdId: str
+    password: str
+    name: str
+    authority: int
+    hide: int = 0
+
+    class Token(BaseModel):
+        access_token: str
+        token_type: str
+        student_id: str
+
+
+class UserInformation(BaseModel):
+    id: int
+    stdId: str
+    name: str
+    authority: int
+
+
+class UserCreate(BaseModel):
+    stdId: str
+    password: str
+    name: str
+
+    @field_validator('stdId', 'password', 'name')
+    def not_empty(cls, v):
+        if not v or not v.strip():
+            raise ValueError('빈 값은 허용되지 않습니다.')
+        return v
+
+    @field_validator('stdId')
+    def student_id_length(cls, v):
+        if len(v) != 8:
+            raise ValueError('학번은 8자리여야 합니다.')
+        return v
+
+
+class UserList(BaseModel):
+    total: int = 0
+    user_list: list[UserInformation] = []
+
+
+class UserUpdate(UserCreate):
+    user_id: int
+
+
+class UserDelete(BaseModel):
+    user_id: int

dependencies.py

@@ -0,0 +1,45 @@
+from enum import Enum
+
+from fastapi import Depends, HTTPException
+from fastapi.security import OAuth2PasswordBearer
+from jose import jwt, JWTError
+from sqlalchemy.orm import Session
+from starlette import status
+from starlette.config import Config
+
+import routers.user.user_crud as user_crud
+from core.database import get_db
+
+config = Config('.env')
+SQLALCHEMY_DATABASE_URL = config('SQLALCHEMY_DATABASE_URL')
+
+ACCESS_TOKEN_EXPIRE_MINUTES = float(config('ACCESS_TOKEN_EXPIRE_MINUTES'))
+SECRET_KEY = config('SECRET_KEY')
+ALGORITHM = config('ALGORITHM')
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/user/login")
+
+
+def get_current_user(token: str = Depends(oauth2_scheme),
+                     db: Session = Depends(get_db)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        student_id: str = payload.get("sub")
+        if student_id is None:
+            raise credentials_exception
+    except JWTError:
+        raise credentials_exception
+    else:
+        student_id = user_crud.get_user(db, student_id=student_id)
+        if student_id is None:
+            raise credentials_exception
+        return student_id
+
+
+class Authority(Enum):
+    Admin = 0
+    User = 1

internal/admin.py

@@ -0,0 +1,55 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from starlette import status
+
+import internal.admin_crud as admin_crud
+from core.database import get_db
+from core.models import models
+from dependencies import get_current_user, Authority
+
+router = APIRouter(
+    prefix="/api/admin",
+    tags=["admin"],
+    dependencies=[Depends(get_current_user)]
+)
+
+
+# Authority 설정
+# Create
+@router.post("/authority", status_code=status.HTTP_204_NO_CONTENT)
+async def create_authority(student_id: str, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
+    if current_user.authority != Authority.Admin.value:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                            detail="권한이 없습니다.")
+
+    admin_crud.set_authority(student_id=student_id, db=db)
+
+
+# Read
+@router.get("/authority/{student_id}")
+async def get_user_authority_level(student_id: str, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
+    if current_user.authority != Authority.Admin.value:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                            detail="권한이 없습니다.")
+
+    return {"authority": Authority(admin_crud.get_authority(student_id=student_id, db=db)).name}
+
+
+# Update
+@router.put("/authority")
+async def toggle_authority(student_id: str, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
+    if current_user.authority != Authority.Admin.value:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                            detail="권한이 없습니다.")
+
+    return {"message": f"Update Authority {admin_crud.toggle_authority(student_id=student_id, db=db)}"}
+
+
+# Delete
+@router.delete("/authority", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_authority(student_id: str, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
+    if current_user.authority != Authority.Admin.value:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                            detail="권한이 없습니다.")
+    admin_crud.unset_authority(student_id=student_id, db=db)
+# !Authority 설정

internal/admin_crud.py

@@ -0,0 +1,29 @@
+from sqlalchemy.orm import Session
+
+from core.models.models import User
+
+
+def set_authority(db: Session, student_id: str):
+    db.query(User).filter(User.stdId == student_id).update({"authority": 0})
+    db.commit()
+
+
+def unset_authority(db: Session, student_id: str):
+    db.query(User).filter(User.stdId == student_id).update({"authority": 1})
+    db.commit()
+
+
+def get_authority(db: Session, student_id: str):
+    return db.query(User).filter(User.stdId == student_id).first().authority
+
+
+def toggle_authority(db: Session, student_id: str):
+    authority = get_authority(db, student_id)
+    if authority == 0:
+        unset_authority(db, student_id)
+    else:
+        set_authority(db, student_id)
+    db.commit()
+
+    return get_authority(db, student_id)
+

main.py

@@ -0,0 +1,28 @@
+from typing import Annotated
+
+import uvicorn
+from fastapi import (
+    Cookie,
+    FastAPI,
+    Query,
+    WebSocket,
+    WebSocketException,
+    WebSocketDisconnect,
+    status,
+)
+from fastapi.responses import HTMLResponse
+
+from internal import admin
+from routers.user import users_router as user
+
+app = FastAPI()
+
+# internal
+app.include_router(admin.router)
+
+# routers
+app.include_router(user.router)
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=80, reload=True)

routers/user/playlist/playlist_crud.py

@@ -0,0 +1,59 @@
+from datetime import datetime
+
+from sqlalchemy.orm import Session
+
+from core.models.models import PlayListInfo, User
+from core.schemas.playlist_info import PlaylistInfoCreate
+from routers.user.user_crud import get_user
+
+
+def create_playlist(db: Session, playlist_create: PlaylistInfoCreate, current_user: User):
+    db_playlist = PlayListInfo(
+        owner=current_user.id,
+        title=playlist_create.title,
+        description=playlist_create.description,
+        create_time=datetime.now(),
+        hide=0,
+    )
+    db.add(db_playlist)
+    db.commit()
+
+    class Config:
+        orm_mode = True
+
+
+def get_playlist_list(db: Session, skip: int = 0, limit: int = 10, keyword: str = ''):
+    _playlist_list = db.query(PlayListInfo).filter(PlayListInfo.hide == 0).order_by(PlayListInfo.id.desc())
+    if keyword:
+        search = '%%{}%%'.format(keyword)
+        _playlist_list = _playlist_list.filter(
+            PlayListInfo.title.ilike(search) |
+            PlayListInfo.description.ilike(search) |
+            PlayListInfo.owner.ilike(search)
+            # | _playlist_list.join(PlayListInfo.playlist).any(PlayListInfo.title.ilike(search))
+        )
+
+    total = _playlist_list.count()
+    playlist_list = _playlist_list.order_by(PlayListInfo.id.desc()).offset(skip).limit(limit).distinct().all()
+
+    return total, playlist_list
+
+
+def get_playlist(db: Session, playlist_id: int):
+    return db.query(PlayListInfo).filter(PlayListInfo.id == playlist_id, PlayListInfo.hide == 0).first()
+
+
+def delete_playlist(db: Session, playlist_id: int):
+    db.query(PlayListInfo).filter(PlayListInfo.id == playlist_id).update({"hide": 1})
+    db.commit()
+
+
+def update_playlist(db: Session, db_playlist: PlayListInfo, playlist_update: PlaylistInfoCreate):
+    db_playlist.title = playlist_update.title
+    db_playlist.description = playlist_update.description
+
+    db.add(db_playlist)
+    db.commit()
+
+    class Config:
+        orm_mode = True