New Branch 생성

.gitignore

@@ -109,6 +109,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+.vscode/
 
 # Spyder project settings
 .spyderproject

README.md

@@ -1,2 +1,30 @@
 <h1 align="center">Cafe Search: CSB</h1>
 <p align="center">1차 프로젝트: 개발자 모각코 분석 프로그램</p>
+
+```
+CAFE-SEARCH-CSB
+├─.gitignore
+├─ README.MD
+├─ requirements.txt
+└─backend/
+  ├─apis/
+  │ ├─version1/
+  │ │ └─ route_users.py
+  │ └─ base.py
+  ├─core/
+  │  └─ config.py
+  ├─db/
+  │  ├─logics/
+  │  │  └─users.py
+  │  ├─models/
+  │  │  └─users.py
+  │  ├─base.py
+  │  ├─base_class.py
+  │  └─session.py
+  ├─schemas/
+  │  └─users.py
+  └─tests/
+     │  └─conftest.py
+     └─test_routes/
+         └─test_users.py
+```

backend/apis/base.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+from apis.version1 import route_users
+from apis.version1 import route_login
+api_router = APIRouter()
+
+api_router.include_router(route_users.router, prefix="/users", tags=["users"])
+api_router.include_router(route_login.router, prefix="/login", tags=["login"])

backend/apis/version1/route_login.py

@@ -0,0 +1,36 @@
+from datetime import timedelta
+
+from core.config import settings
+from db.session import get_db
+from db.logics.users import get_user
+from db.logics.login import create_access_token
+from fastapi import (
+    APIRouter,
+    Depends,
+    HTTPException,
+    status
+)
+from fastapi.security import OAuth2PasswordRequestForm
+
+from schemas.tokens import Token
+from sqlalchemy.orm import Session
+
+router = APIRouter()
+
+@router.post("/token", response_model=Token)
+def login_for_access_token(
+    form_data: OAuth2PasswordRequestForm = Depends(),
+    db: Session = Depends(get_db)
+):
+    user = get_user(form_data.username, form_data.password, db)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect Username or Password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"user_email":user.email}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}

backend/apis/version1/route_users.py

@@ -0,0 +1,13 @@
+from db.session import get_db
+from db.logics.users import create_new_user
+from fastapi import APIRouter, Depends, status
+from schemas.users import UserCreate, ShowUser
+from sqlalchemy.orm import Session
+
+router = APIRouter()
+
+
+@router.post("/", response_model=ShowUser, status_code=status.HTTP_201_CREATED)
+def create_user(user: UserCreate, db: Session = Depends(get_db)):
+    user = create_new_user(user=user, db=db)
+    return user

backend/core/config.py

@@ -0,0 +1,45 @@
+from pydantic import BaseSettings, SecretStr
+
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parents[2]
+
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "Cafe Search"
+    PROJECT_VERSION: str = "1.0.0"
+
+    DB_USERNAME: str
+    DB_PASSWORD: SecretStr
+    DB_HOST: str
+    DB_PORT: int
+    DB_NAME: str
+
+    SECRET_KEY: str
+    SECRET_ALGORITHM = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+
+    class Config:
+        env_file = str(BASE_DIR / ".env")
+        env_file_encoding = "utf-8"
+
+class TestSettings(BaseSettings):
+    TEST_DB_USERNAME: str
+    TEST_DB_PASSWORD: SecretStr
+    TEST_DB_HOST: str
+    TEST_DB_PORT: int
+    TEST_DB_NAME: str
+
+    SECRET_KEY: str
+    SECRET_ALGORITHM = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+
+    class Config:
+        env_file = str(BASE_DIR / ".env")
+        env_file_encoding = "utf-8"
+
+
+
+settings = Settings()

backend/core/hashing.py

@@ -0,0 +1,12 @@
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+class Hasher:
+    @staticmethod
+    def verify_password(plain_password, hashed_password):
+        return pwd_context.verify(plain_password, hashed_password)
+
+    @staticmethod
+    def get_password_hash(password):
+        return pwd_context.hash(password)

backend/db/base.py

@@ -0,0 +1,2 @@
+from db.base_class import Base
+from db.models.users import User

backend/db/base_class.py

@@ -0,0 +1,12 @@
+from typing import Any
+from sqlalchemy.ext.declarative import as_declarative, declared_attr
+
+
+@as_declarative()
+class Base:
+    id: Any
+    __name__: str
+
+    @declared_attr
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()

backend/db/logics/login.py

@@ -0,0 +1,18 @@
+from core.config import settings
+from datetime import datetime, timedelta
+from jose import jwt
+from typing import Optional
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta]=None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(
+            minutes = settings.ACCESS_TOKEN_EXPIRE_MINUTES
+        )
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(
+        to_encode, settings.SECRET_KEY, algorithm=settings.SECRET_ALGORITHM
+    )
+    return encoded_jwt

backend/db/logics/users.py

@@ -0,0 +1,25 @@
+from core.hashing import Hasher
+from db.models.users import User
+from schemas.users import UserCreate
+from sqlalchemy.orm import Session
+
+
+def create_new_user(user: UserCreate, db: Session) -> User:
+    user = User(
+        username=user.username,
+        email=user.email,
+        hashed_password=Hasher.get_password_hash(user.password),
+        is_superuser=False,
+    )
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+def get_user(username: str, password: str, db: Session) -> User:
+    user = db.query(User).filter(User.email == username).first()
+    if not user:
+        return False
+    if not Hasher.verify_password(password, user.hashed_password):
+        return False
+    return user

backend/db/models/base.py

@@ -0,0 +1,6 @@
+from sqlalchemy import Column, Integer, DateTime, func
+
+class BaseMixin:
+    id = Column(Integer, primary_key=True, index=True)
+    created_at = Column(DateTime, nullable=False, default=func.utc_timestamp())
+    updated_at = Column(DateTime, nullable=False, default=func.utc_timestamp(), onupdate=func.utc_timestamp())

backend/db/models/users.py

@@ -0,0 +1,11 @@
+from sqlalchemy import Column, String, Boolean
+
+from db.base_class import Base
+from db.models.base import BaseMixin
+
+
+class User(Base, BaseMixin):
+    username = Column(String(40), unique=True, nullable=False)
+    email = Column(String(60), unique=True, nullable=False, index=True)
+    hashed_password = Column(String(100), nullable=False)
+    is_superuser = Column(Boolean(), default=False)

backend/db/session.py

@@ -0,0 +1,25 @@
+from typing import Generator
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from core.config import settings
+
+engine = create_engine(
+    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}?charset=utf8mb4".format(
+        username=settings.DB_USERNAME,
+        password=settings.DB_PASSWORD.get_secret_value(),
+        host=settings.DB_HOST,
+        port=settings.DB_PORT,
+        name=settings.DB_NAME,
+    )
+)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+def get_db() -> Generator:
+    try:
+        db = SessionLocal()
+        yield db
+    finally:
+        db.close()

backend/main.py

@@ -0,0 +1,16 @@
+from apis.base import api_router
+from core.config import settings
+from fastapi import FastAPI
+
+
+def include_router(app: FastAPI):
+    app.include_router(api_router)
+
+
+def start_application():
+    app = FastAPI(title=settings.PROJECT_NAME, version=settings.PROJECT_VERSION)
+    include_router(app)
+    return app
+
+
+app = start_application()

backend/schemas/tokens.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel, EmailStr
+
+class UserCreate(BaseModel):
+    username: str
+    email: EmailStr
+    password: str
+
+class ShowUser(BaseModel):
+    username: str
+    email: EmailStr
+
+    class Config:
+        orm_mode = True

backend/schemas/users.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str

backend/tests/conftest.py

@@ -0,0 +1,65 @@
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from typing import Any, Generator
+
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from apis.base import api_router
+from db.base import Base
+from db.session import get_db
+from core.config import TestSettings
+
+def start_application() -> FastAPI:
+    app = FastAPI()
+    app.include_router(api_router)
+    return app
+
+
+settings = TestSettings()
+engine = create_engine(
+    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}?charset=utf8mb4".format(
+        username=settings.TEST_DB_USERNAME,
+        password=settings.TEST_DB_PASSWORD.get_secret_value(),
+        host=settings.TEST_DB_HOST,
+        port=settings.TEST_DB_PORT,
+        name=settings.TEST_DB_NAME,
+    )
+)
+
+SessionTesting = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+@pytest.fixture(scope="module")
+def app() -> Generator[FastAPI, Any, None]:
+    Base.metadata.create_all(engine)
+    _app = start_application()
+    yield _app
+    Base.metadata.drop_all(engine)
+
+@pytest.fixture(scope="module")
+def db_session(app: FastAPI) -> Generator[SessionTesting, Any, None]:
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = SessionTesting(bind=connection)
+    yield session
+    session.close()
+    transaction.rollback()
+    connection.close()
+
+@pytest.fixture(scope="module")
+def client(
+    app: FastAPI, db_session: SessionTesting
+) -> Generator[TestClient, Any, None]:
+    def _get_test_db():
+        try:
+            yield db_session
+        finally:
+            pass
+
+    app.dependency_overrides[get_db] = _get_test_db
+    with TestClient(app) as client:
+        yield client

backend/tests/test_routes/test_users.py

@@ -0,0 +1,12 @@
+import json
+
+def test_create_user(client):
+    data = {
+        "username": "test-sbjo",
+        "email": "testsb@jo.com",
+        "password" : "test-password"
+    }
+    response = client.post("/users/", json.dumps(data))
+    assert response.status_code == 201
+    assert response.json()["username"] == "test-sbjo"
+    assert response.json()["email"] == "testsb@jo.com"

requirements.txt

@@ -0,0 +1,10 @@
+pymysql
+pydantic[dotenv]
+
+sqlalchemy
+fastapi
+requests
+pytest
+uvicorn[standard]
+python-jose[cryptography]
+passlib[bcrypt]