first commit

app/__init__.py

@@ -0,0 +1,30 @@
+from fastapi import FastAPI, Depends, HTTPException, status, Request, Response
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from jose import JWTError, jwt
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.orm import Session
+
+from app import models
+from app import schemas
+from app import database
+from app import crud
+from app.utils import encrypt
+from app.api import api_router
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+database.Base.metadata.create_all(bind=database.engine)
+app = FastAPI()
+
+
+@app.middleware("http")
+async def db_session_middleware(request: Request, call_next):
+    response = Response("Internal server error", status_code=500)
+    try:
+        request.state.db = database.SessionLocal()
+        response = await call_next(request)
+    finally:
+        request.state.db.close()
+    return response
+
+
+app.include_router(api_router, prefix="/api")

app/api/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from app.api import user
+
+api_router = APIRouter()
+api_router.include_router(user.router, tags=["user"])

app/api/user.py

@@ -0,0 +1,67 @@
+from fastapi import (
+    FastAPI,
+    Depends,
+    HTTPException,
+    status,
+    Request,
+    Response,
+    APIRouter,
+)
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from jose import JWTError, jwt
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.orm import Session
+from typing import List
+
+from app import models
+from app import schemas
+from app import database
+from app import crud
+from app.utils import encrypt
+from app.utils import depes
+
+
+router = APIRouter()
+
+
+def get_db(request: Request):
+    return request.state.db
+
+
+@router.post("/user/create/", response_model=schemas.User)
+async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
+    db_user = crud.crud_user.get_user(db, username=user.username)
+    if db_user:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Username already registered",
+        )
+    return crud.crud_user.create_user(db=db, user=user)
+
+
+@router.post("/auth/token", response_model=schemas.Token)
+async def create_token(
+    form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)
+):
+    user = crud.crud_user.authenticate_user(
+        db, username=form.username, password=form.password
+    )
+
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+        )
+
+    access_token = encrypt.create_token(data={"username": user.username})
+
+    return {
+        "access_token": access_token,
+        "token_type": "bearer",
+    }
+
+
+@router.get("/user/", response_model=List[schemas.User])
+def get_users(db: Session = Depends(get_db), token = Depends(depes.get_current_user)):
+    return crud.crud_user.get_users(db)
+

app/crud/__init__.py

@@ -0,0 +1 @@
+from app.crud import crud_user

app/crud/crud_user.py

@@ -0,0 +1,34 @@
+from sqlalchemy.orm import Session
+
+from app import models, schemas
+from app.utils.encrypt import get_password_hash, verify_password
+
+
+def get_user(db: Session, username: str):
+    return db.query(models.User).filter(models.User.username == username).first()
+
+
+def create_user(db: Session, user: schemas.UserCreate):
+    hashed_password = get_password_hash(user.password)
+    user = models.User(
+        username=user.username,
+        email=user.email,
+        full_name=user.full_name,
+        hashed_password=hashed_password,
+    )
+    db.add(user)
+    db.commit()
+    return user
+
+
+def get_users(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.User).offset(skip).limit(limit).all()
+
+
+def authenticate_user(db: Session, username: str, password: str):
+    user = get_user(db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.hashed_password):
+        return False
+    return user

app/database.py

@@ -0,0 +1,9 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import declarative_base
+
+
+DATABASE_URL = "mysql+pymysql://root:example@localhost:3306/wncloud?charset=utf8mb4"
+engine = create_engine(DATABASE_URL)
+SessionLocal = sessionmaker(autoflush=False, bind=engine)
+Base = declarative_base()

app/models/__init__.py

@@ -0,0 +1 @@
+from .user import User

app/models/user.py

@@ -0,0 +1,22 @@
+import uuid
+
+from app.database import Base
+
+
+from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
+
+def gen_uuid():
+    return str(uuid.uuid4())
+
+
+class User(Base):
+    __tablename__ = "user"
+
+    id = Column(Integer, primary_key=True, index=True)
+    uuid = Column(String(50), unique=True, default=gen_uuid)
+    username = Column(String(50), unique=True, index=True)
+    full_name = Column(String(50))
+    email = Column(String(50), unique=True, index=True)
+    hashed_password = Column(String(64))
+    deleted = Column(Integer, default=0)
+

app/schemas/__init__.py

@@ -0,0 +1 @@
+from .user import UserBase, UserCreate, User, Token

app/schemas/user.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+
+class UserBase(BaseModel):
+    username: str
+    email: Optional[str] = None
+    full_name: Optional[str] = None
+
+
+class UserCreate(UserBase):
+    password: str
+
+
+class User(UserBase):
+    class Config:
+        from_attributes = True

app/utils/__init__.py

@@ -0,0 +1 @@
+from app.utils import encrypt

app/utils/depes.py

@@ -0,0 +1,35 @@
+import jwt
+
+from sqlalchemy.orm import Session
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+
+from app import models, schemas, crud
+from app.utils.encrypt import decode_token
+from app.database import SessionLocal
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token")
+
+
+def get_db():
+    try:
+        db = SessionLocal()
+        yield db
+    finally:
+        db.close()
+
+def get_current_user(
+    db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
+) -> models.User:
+    try:
+        payload = decode_token(token)
+        print(payload)
+    except jwt.JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Could not validate credentials",
+        )
+    user = crud.crud_user.get_user(db, username=payload.get("username"))
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return user

app/utils/encrypt.py

@@ -0,0 +1,31 @@
+from datetime import timedelta, datetime
+from passlib.context import CryptContext
+from jose import jwt
+
+SECRET_KEY = "6ed23d121dd868fd6c0bf3fce00140ccc196f162f842fda574901052f2127e09"
+ACCESS_TOKEN_EXPIRE_MINUTES = 15
+AlGORITHM = "HS256"
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+
+def create_token(data: dict):
+    to_encode = data.copy()
+    expires_delta = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    expire = datetime.now() + expires_delta
+    to_encode.update({"exp": expire})
+    encode_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=AlGORITHM)
+    return encode_jwt
+
+
+def decode_token(token):
+    payload = jwt.decode(token, SECRET_KEY, algorithms=[AlGORITHM])
+    return payload