mirror of
https://github.com/maxdorninger/MediaManager.git
synced 2026-04-17 21:54:00 +02:00
add comments and remove unused imports
This commit is contained in:
@@ -3,14 +3,13 @@ from typing import Annotated
|
||||
|
||||
import bcrypt
|
||||
import jwt
|
||||
from fastapi import Depends, FastAPI, HTTPException, status, APIRouter
|
||||
from fastapi import Depends, HTTPException, status, APIRouter
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from jwt.exceptions import InvalidTokenError
|
||||
from passlib.context import CryptContext
|
||||
from pydantic import BaseModel
|
||||
|
||||
import database
|
||||
from database import UserInternal, User
|
||||
from database import UserInternal
|
||||
|
||||
# to get a string like this run:
|
||||
# openssl rand -hex 32
|
||||
@@ -18,7 +17,6 @@ SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
||||
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
@@ -58,13 +56,19 @@ def get_password_hash(password):
|
||||
)
|
||||
|
||||
|
||||
def authenticate_user(email: str, password: str) -> UserInternal:
|
||||
def authenticate_user(email: str, password: str) -> bool | UserInternal:
|
||||
"""
|
||||
|
||||
:param email: email of the user
|
||||
:param password: password of the user
|
||||
:return: if authentication succeeds, returns the user object with added name and lastname, otherwise or if the user doesn't exist returns False
|
||||
"""
|
||||
user = database.get_user(email)
|
||||
if not user:
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
return user
|
||||
return True
|
||||
|
||||
|
||||
def create_access_token(data: dict, expires_delta: timedelta | None = None):
|
||||
@@ -97,11 +101,12 @@ async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
|
||||
raise credentials_exception
|
||||
return user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login_for_access_token(
|
||||
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
|
||||
) -> Token:
|
||||
print("post:",form_data.username, form_data.password)
|
||||
print("post:", form_data.username, form_data.password)
|
||||
user = authenticate_user(form_data.username, form_data.password)
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
@@ -113,4 +118,4 @@ async def login_for_access_token(
|
||||
access_token = create_access_token(
|
||||
data={"sub": user.email}, expires_delta=access_token_expires
|
||||
)
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from abc import ABC, abstractmethod
|
||||
from logging import getLogger
|
||||
from uuid import uuid4
|
||||
@@ -13,12 +12,18 @@ log.level = logging.DEBUG
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
"""
|
||||
User model
|
||||
"""
|
||||
name: str
|
||||
lastname: str
|
||||
email: str
|
||||
|
||||
|
||||
class UserInternal(User):
|
||||
""""
|
||||
Internal user model, assumes the password is already hashed, when a new instance is created
|
||||
"""
|
||||
id: str = str(uuid4())
|
||||
hashed_password: str
|
||||
|
||||
@@ -81,6 +86,11 @@ def drop_tables() -> None:
|
||||
|
||||
|
||||
def create_user(user: UserInternal) -> bool:
|
||||
"""
|
||||
|
||||
:param user: user to create, password must already be hashed
|
||||
:return: True if user was created, False otherwise
|
||||
"""
|
||||
with PgDatabase() as db:
|
||||
try:
|
||||
db.connection.execute(
|
||||
@@ -100,6 +110,11 @@ def create_user(user: UserInternal) -> bool:
|
||||
|
||||
|
||||
def get_user(email: str) -> UserInternal | None:
|
||||
"""
|
||||
|
||||
:param email: the users email address
|
||||
:return: if user was found its is returned, otherwise None
|
||||
"""
|
||||
with PgDatabase() as db:
|
||||
result = db.connection.execute(
|
||||
"SELECT id, name, lastname, email, hashed_password FROM users WHERE email=%s",
|
||||
|
||||
@@ -1,18 +1,12 @@
|
||||
import logging
|
||||
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, Depends
|
||||
from pydantic import BaseModel
|
||||
from fastapi import FastAPI
|
||||
|
||||
import database
|
||||
from fastapi.testclient import TestClient
|
||||
from routers import users
|
||||
from auth import password
|
||||
from routers.users import CreateUser
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
logging.info("OIDA")
|
||||
app.include_router(users.router, tags=["users"])
|
||||
app.include_router(password.app, tags=["authentication"])
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@ import logging
|
||||
|
||||
from fastapi import APIRouter
|
||||
from fastapi import Depends
|
||||
from pydantic import BaseModel
|
||||
|
||||
import database
|
||||
from auth.password import authenticate_user, get_password_hash
|
||||
@@ -14,6 +13,9 @@ router = APIRouter(
|
||||
|
||||
|
||||
class CreateUser(User):
|
||||
""""
|
||||
The Usermodel, but with an additional non-hashed password. attribute
|
||||
"""
|
||||
password: str
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
Reference in New Issue
Block a user