Fix import

.env.sample

@@ -0,0 +1,4 @@
+# openssl rand -hex 32
+SECRET_KEY=
+ALGORITHM=HS256
+ACCESS_TOKEN_EXPIRE_MINUTES=30

.gitignore

@@ -3,6 +3,7 @@
 __pycache__/
 *.py[cod]
 *$py.class
+requirements.txt
 
 # C extensions
 *.so

Dockerfile

@@ -1,6 +1,6 @@
 FROM tiangolo/uvicorn-gunicorn-fastapi:python3.8-slim
 ENV DOCKER=1
 
-COPY ./requirements.txt /
-RUN pip install -r /requirements.txt && rm /requirements.txt
+COPY requirements.txt /app/requirements.txt
+RUN pip install -r /app/requirements.txt 
 COPY ./app /app/app

Pipfile

@@ -7,13 +7,16 @@ name = "pypi"
 fastapi = "*"
 fastapi-responses = "*"
 uvicorn = {extras = ["standard"], version = "*"}
+python-jose = {extras = ["cryptography"], version = "*"}
+passlib = "*"
+python-multipart = "*"
 
 [dev-packages]
-# pytest = "*"
-# requests = "*"
-# pytest-asyncio = "*"
-# black = "*"
-# mypy = "*"
+pytest = "*"
+requests = "*"
+pytest-asyncio = "*"
+black = "*"
+mypy = "*"
 
 [requires]
 python_version = "3.8"

README.md

@@ -1,5 +1,4 @@
-# FastApi Template
-
+# FastAPI Oauth2
 ```pipenv sync```
 
 ```pipenv sync --dev```

app/auth.py

@@ -0,0 +1,81 @@
+from os import getenv
+import json
+from datetime import timedelta, datetime
+from typing import Optional
+from fastapi.security import OAuth2PasswordBearer
+from fastapi import Depends, HTTPException, status
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from app.models import User, UserInDB, TokenData
+
+
+# to get a string like this run:
+# openssl rand -hex 32
+SECRET_KEY = getenv("SECRET_KEY")
+ALGORITHM = getenv("ALGORITHM")
+ACCESS_TOKEN_EXPIRE_MINUTES = int(getenv("ACCESS_TOKEN_EXPIRE_MINUTES"))
+
+fake_users_db = json.load(open("app/users.json"))
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+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 get_user(db, username: str):
+    if username in db:
+        user_dict = db[username]
+        return UserInDB(**user_dict)
+
+
+def authenticate_user(fake_db, username: str, password: str):
+    user = get_user(fake_db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.hashed_password):
+        return False
+    return user
+
+
+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=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    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])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    user = get_user(fake_users_db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+
+async def get_current_active_user(current_user: User = Depends(get_current_user)):
+    if current_user.disabled:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    return current_user

app/main.py

@@ -1,117 +1,14 @@
-from datetime import datetime, timedelta
-from typing import Optional
+from datetime import timedelta
+
 
 from fastapi import Depends, FastAPI, HTTPException, status
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from jose import JWTError, jwt
-from passlib.context import CryptContext
-from pydantic import BaseModel
+from fastapi.security import OAuth2PasswordRequestForm
 
-# to get a string like this run:
-# openssl rand -hex 32
-SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 30
-
-
-fake_users_db = {
-    "johndoe": {
-        "username": "johndoe",
-        "full_name": "John Doe",
-        "email": "johndoe@example.com",
-        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
-        "disabled": False,
-    }
-}
-
-
-class Token(BaseModel):
-    access_token: str
-    token_type: str
-
-
-class TokenData(BaseModel):
-    username: Optional[str] = None
-
-
-class User(BaseModel):
-    username: str
-    email: Optional[str] = None
-    full_name: Optional[str] = None
-    disabled: Optional[bool] = None
-
-
-class UserInDB(User):
-    hashed_password: str
-
-
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+from app.models import Token, User
+from app.auth import get_current_active_user, create_access_token, authenticate_user, fake_users_db, ACCESS_TOKEN_EXPIRE_MINUTES
 
 app = FastAPI()
 
-
-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 get_user(db, username: str):
-    if username in db:
-        user_dict = db[username]
-        return UserInDB(**user_dict)
-
-
-def authenticate_user(fake_db, username: str, password: str):
-    user = get_user(fake_db, username)
-    if not user:
-        return False
-    if not verify_password(password, user.hashed_password):
-        return False
-    return user
-
-
-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=15)
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
-
-
-async def get_current_user(token: str = Depends(oauth2_scheme)):
-    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])
-        username: str = payload.get("sub")
-        if username is None:
-            raise credentials_exception
-        token_data = TokenData(username=username)
-    except JWTError:
-        raise credentials_exception
-    user = get_user(fake_users_db, username=token_data.username)
-    if user is None:
-        raise credentials_exception
-    return user
-
-
-async def get_current_active_user(current_user: User = Depends(get_current_user)):
-    if current_user.disabled:
-        raise HTTPException(status_code=400, detail="Inactive user")
-    return current_user
-
-
 @app.post("/token", response_model=Token)
 async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
     user = authenticate_user(fake_users_db, form_data.username, form_data.password)

app/models.py

@@ -0,0 +1,20 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(BaseModel):
+    username: Optional[str] = None
+
+
+class User(BaseModel):
+    username: str
+    email: Optional[str] = None
+    full_name: Optional[str] = None
+    disabled: Optional[bool] = None
+
+class UserInDB(User):
+    hashed_password: str

app/users.json

@@ -0,0 +1,9 @@
+{
+    "johndoe": {
+        "username": "johndoe",
+        "full_name": "John Doe",
+        "email": "johndoe@example.com",
+        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
+        "disabled": "False"
+    }
+}

docker-compose-dev.yml

@@ -0,0 +1,14 @@
+version: '3.7'
+
+services:
+    app:
+        build: .
+        env_file: 
+            - .env
+        ports: 
+            - 8000:8000
+        tty: true
+        stdin_open: true
+        volumes:
+          - ./app:/app/app
+        command: uvicorn app.main:app --host 0.0.0.0 --reload

docker-compose-test.yml

@@ -0,0 +1,14 @@
+version: '3.7'
+
+services:
+    app:
+        build: .
+        env_file: 
+            - .env
+        ports: 
+            - 8000:80
+        tty: true
+        stdin_open: true
+        volumes:
+          - ./app:/app/app
+        command: pytest app/test/test_main.py -s

requirements.txt

@@ -1,16 +0,0 @@
--i https://pypi.org/simple
-asgiref==3.4.1; python_version >= '3.6'
-click==8.0.1; python_version >= '3.6'
-fastapi-responses==0.2.1
-fastapi==0.68.1
-h11==0.12.0; python_version >= '3.6'
-httptools==0.2.0
-pydantic==1.8.2; python_full_version >= '3.6.1'
-python-dotenv==0.19.0
-pyyaml==5.4.1
-starlette==0.14.2; python_version >= '3.6'
-typing-extensions==3.10.0.2
-uvicorn[standard]==0.15.0
-uvloop==0.16.0
-watchgod==0.7
-websockets==10.0