klinsmaya/SQLBot
1
0
Fork 0
mirror of https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git synced 2026-11-08 01:06:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: Login api

Browse Source
This commit is contained in:
fit2cloud-chenyw
2025-04-23 21:14:42 +08:00
parent 6b084f0865
commit 051507504b
56 changed files with 213 additions and 1900 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/alembic.ini
+1 -1
View File
@@ -2,7 +2,7 @@
[alembic]
# path to migration scripts
script_location = app/alembic
script_location = alembic
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
backend/app/alembic/README → backend/alembic/README
View File
backend/app/alembic/env.py → backend/alembic/env.py
+2 -2
View File
@@ -18,8 +18,8 @@ fileConfig(config.config_file_name)
# target_metadata = mymodel.Base.metadata
# target_metadata = None
from app.models import SQLModel # noqa
from app.core.config import settings # noqa
from apps.system.models.user import SQLModel # noqa
from common.core.config import settings # noqa
target_metadata = SQLModel.metadata
backend/app/alembic/script.py.mako → backend/alembic/script.py.mako
View File
backend/app/alembic/versions/.keep → backend/alembic/versions/.keep
View File
backend/app/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py → backend/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py
View File
backend/app/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py → backend/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py
View File
backend/app/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py → backend/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py
View File
backend/app/alembic/versions/e2412789c190_initialize_models.py → backend/alembic/versions/e2412789c190_initialize_models.py
View File
backend/app/api/deps.py
-57
View File
@@ -1,57 +0,0 @@
from collections.abc import Generator
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pydantic import ValidationError
from sqlmodel import Session
from app.core import security
from app.core.config import settings
from app.core.db import engine
from app.models import TokenPayload, User
reusable_oauth2 = OAuth2PasswordBearer(
tokenUrl=f"{settings.API_V1_STR}/login/access-token"
)
def get_db() -> Generator[Session, None, None]:
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_db)]
TokenDep = Annotated[str, Depends(reusable_oauth2)]
def get_current_user(session: SessionDep, token: TokenDep) -> User:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
)
token_data = TokenPayload(**payload)
except (InvalidTokenError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = session.get(User, token_data.sub)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return user
CurrentUser = Annotated[User, Depends(get_current_user)]
def get_current_active_superuser(current_user: CurrentUser) -> User:
if not current_user.is_superuser:
raise HTTPException(
status_code=403, detail="The user doesn't have enough privileges"
)
return current_user
backend/app/api/main.py
-14
View File
@@ -1,14 +0,0 @@
from fastapi import APIRouter
from app.api.routes import items, login, private, users, utils
from app.core.config import settings
api_router = APIRouter()
api_router.include_router(login.router)
api_router.include_router(users.router)
api_router.include_router(utils.router)
api_router.include_router(items.router)
if settings.ENVIRONMENT == "local":
api_router.include_router(private.router)
backend/app/api/routes/items.py
-109
View File
@@ -1,109 +0,0 @@
import uuid
from typing import Any
from fastapi import APIRouter, HTTPException
from sqlmodel import func, select
from app.api.deps import CurrentUser, SessionDep
from app.models import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate, Message
router = APIRouter(prefix="/items", tags=["items"])
@router.get("/", response_model=ItemsPublic)
def read_items(
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
) -> Any:
"""
Retrieve items.
"""
if current_user.is_superuser:
count_statement = select(func.count()).select_from(Item)
count = session.exec(count_statement).one()
statement = select(Item).offset(skip).limit(limit)
items = session.exec(statement).all()
else:
count_statement = (
select(func.count())
.select_from(Item)
.where(Item.owner_id == current_user.id)
)
count = session.exec(count_statement).one()
statement = (
select(Item)
.where(Item.owner_id == current_user.id)
.offset(skip)
.limit(limit)
)
items = session.exec(statement).all()
return ItemsPublic(data=items, count=count)
@router.get("/{id}", response_model=ItemPublic)
def read_item(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
"""
Get item by ID.
"""
item = session.get(Item, id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
if not current_user.is_superuser and (item.owner_id != current_user.id):
raise HTTPException(status_code=400, detail="Not enough permissions")
return item
@router.post("/", response_model=ItemPublic)
def create_item(
*, session: SessionDep, current_user: CurrentUser, item_in: ItemCreate
) -> Any:
"""
Create new item.
"""
item = Item.model_validate(item_in, update={"owner_id": current_user.id})
session.add(item)
session.commit()
session.refresh(item)
return item
@router.put("/{id}", response_model=ItemPublic)
def update_item(
*,
session: SessionDep,
current_user: CurrentUser,
id: uuid.UUID,
item_in: ItemUpdate,
) -> Any:
"""
Update an item.
"""
item = session.get(Item, id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
if not current_user.is_superuser and (item.owner_id != current_user.id):
raise HTTPException(status_code=400, detail="Not enough permissions")
update_dict = item_in.model_dump(exclude_unset=True)
item.sqlmodel_update(update_dict)
session.add(item)
session.commit()
session.refresh(item)
return item
@router.delete("/{id}")
def delete_item(
session: SessionDep, current_user: CurrentUser, id: uuid.UUID
) -> Message:
"""
Delete an item.
"""
item = session.get(Item, id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
if not current_user.is_superuser and (item.owner_id != current_user.id):
raise HTTPException(status_code=400, detail="Not enough permissions")
session.delete(item)
session.commit()
return Message(message="Item deleted successfully")
backend/app/api/routes/login.py
-72
View File
@@ -1,72 +0,0 @@
from datetime import timedelta
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.security import OAuth2PasswordRequestForm
from app import crud
from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
from app.core import security
from app.core.config import settings
from app.core.security import get_password_hash
from app.models import Message, NewPassword, Token, UserPublic
from app.utils import (
generate_password_reset_token,
verify_password_reset_token,
)
router = APIRouter(tags=["login"])
@router.post("/login/access-token")
def login_access_token(
session: SessionDep, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
) -> Token:
"""
OAuth2 compatible token login, get an access token for future requests
"""
user = crud.authenticate(
session=session, email=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email or password")
elif not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return Token(
access_token=security.create_access_token(
user.id, expires_delta=access_token_expires
)
)
@router.post("/login/test-token", response_model=UserPublic)
def test_token(current_user: CurrentUser) -> Any:
"""
Test access token
"""
return current_user
@router.post("/reset-password/")
def reset_password(session: SessionDep, body: NewPassword) -> Message:
"""
Reset password
"""
email = verify_password_reset_token(token=body.token)
if not email:
raise HTTPException(status_code=400, detail="Invalid token")
user = crud.get_user_by_email(session=session, email=email)
if not user:
raise HTTPException(
status_code=404,
detail="The user with this email does not exist in the system.",
)
elif not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
hashed_password = get_password_hash(password=body.new_password)
user.hashed_password = hashed_password
session.add(user)
session.commit()
return Message(message="Password updated successfully")
backend/app/api/routes/private.py
-38
View File
@@ -1,38 +0,0 @@
from typing import Any
from fastapi import APIRouter
from pydantic import BaseModel
from app.api.deps import SessionDep
from app.core.security import get_password_hash
from app.models import (
User,
UserPublic,
)
router = APIRouter(tags=["private"], prefix="/private")
class PrivateUserCreate(BaseModel):
email: str
password: str
full_name: str
is_verified: bool = False
@router.post("/users/", response_model=UserPublic)
def create_user(user_in: PrivateUserCreate, session: SessionDep) -> Any:
"""
Create a new user.
"""
user = User(
email=user_in.email,
full_name=user_in.full_name,
hashed_password=get_password_hash(user_in.password),
)
session.add(user)
session.commit()
return user
backend/app/api/routes/users.py
-199
View File
@@ -1,199 +0,0 @@
import uuid
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import col, delete, func, select
from app import crud
from app.api.deps import (
CurrentUser,
SessionDep,
get_current_active_superuser,
)
from app.core.config import settings
from app.core.security import get_password_hash, verify_password
from app.models import (
Item,
Message,
UpdatePassword,
User,
UserCreate,
UserPublic,
UserRegister,
UsersPublic,
UserUpdate,
UserUpdateMe,
)
router = APIRouter(prefix="/users", tags=["users"])
@router.get(
"/",
dependencies=[Depends(get_current_active_superuser)],
response_model=UsersPublic,
)
def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
"""
Retrieve users.
"""
count_statement = select(func.count()).select_from(User)
count = session.exec(count_statement).one()
statement = select(User).offset(skip).limit(limit)
users = session.exec(statement).all()
return UsersPublic(data=users, count=count)
@router.patch("/me", response_model=UserPublic)
def update_user_me(
*, session: SessionDep, user_in: UserUpdateMe, current_user: CurrentUser
) -> Any:
"""
Update own user.
"""
if user_in.email:
existing_user = crud.get_user_by_email(session=session, email=user_in.email)
if existing_user and existing_user.id != current_user.id:
raise HTTPException(
status_code=409, detail="User with this email already exists"
)
user_data = user_in.model_dump(exclude_unset=True)
current_user.sqlmodel_update(user_data)
session.add(current_user)
session.commit()
session.refresh(current_user)
return current_user
@router.patch("/me/password", response_model=Message)
def update_password_me(
*, session: SessionDep, body: UpdatePassword, current_user: CurrentUser
) -> Any:
"""
Update own password.
"""
if not verify_password(body.current_password, current_user.hashed_password):
raise HTTPException(status_code=400, detail="Incorrect password")
if body.current_password == body.new_password:
raise HTTPException(
status_code=400, detail="New password cannot be the same as the current one"
)
hashed_password = get_password_hash(body.new_password)
current_user.hashed_password = hashed_password
session.add(current_user)
session.commit()
return Message(message="Password updated successfully")
@router.get("/me", response_model=UserPublic)
def read_user_me(current_user: CurrentUser) -> Any:
"""
Get current user.
"""
return current_user
@router.delete("/me", response_model=Message)
def delete_user_me(session: SessionDep, current_user: CurrentUser) -> Any:
"""
Delete own user.
"""
if current_user.is_superuser:
raise HTTPException(
status_code=403, detail="Super users are not allowed to delete themselves"
)
session.delete(current_user)
session.commit()
return Message(message="User deleted successfully")
@router.post("/signup", response_model=UserPublic)
def register_user(session: SessionDep, user_in: UserRegister) -> Any:
"""
Create new user without the need to be logged in.
"""
user = crud.get_user_by_email(session=session, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="The user with this email already exists in the system",
)
user_create = UserCreate.model_validate(user_in)
user = crud.create_user(session=session, user_create=user_create)
return user
@router.get("/{user_id}", response_model=UserPublic)
def read_user_by_id(
user_id: uuid.UUID, session: SessionDep, current_user: CurrentUser
) -> Any:
"""
Get a specific user by id.
"""
user = session.get(User, user_id)
if user == current_user:
return user
if not current_user.is_superuser:
raise HTTPException(
status_code=403,
detail="The user doesn't have enough privileges",
)
return user
@router.patch(
"/{user_id}",
dependencies=[Depends(get_current_active_superuser)],
response_model=UserPublic,
)
def update_user(
*,
session: SessionDep,
user_id: uuid.UUID,
user_in: UserUpdate,
) -> Any:
"""
Update a user.
"""
db_user = session.get(User, user_id)
if not db_user:
raise HTTPException(
status_code=404,
detail="The user with this id does not exist in the system",
)
if user_in.email:
existing_user = crud.get_user_by_email(session=session, email=user_in.email)
if existing_user and existing_user.id != user_id:
raise HTTPException(
status_code=409, detail="User with this email already exists"
)
db_user = crud.update_user(session=session, db_user=db_user, user_in=user_in)
return db_user
@router.delete("/{user_id}", dependencies=[Depends(get_current_active_superuser)])
def delete_user(
session: SessionDep, current_user: CurrentUser, user_id: uuid.UUID
) -> Message:
"""
Delete a user.
"""
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if user == current_user:
raise HTTPException(
status_code=403, detail="Super users are not allowed to delete themselves"
)
statement = delete(Item).where(col(Item.owner_id) == user_id)
session.exec(statement) # type: ignore
session.delete(user)
session.commit()
return Message(message="User deleted successfully")
backend/app/api/routes/utils.py
-14
View File
@@ -1,14 +0,0 @@
from fastapi import APIRouter, Depends
from pydantic.networks import EmailStr
from app.api.deps import get_current_active_superuser
from app.models import Message
router = APIRouter(prefix="/utils", tags=["utils"])
@router.get("/health-check/")
async def health_check() -> bool:
return True
backend/app/backend_pre_start.py
-39
View File
@@ -1,39 +0,0 @@
import logging
from sqlalchemy import Engine
from sqlmodel import Session, select
from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed
from app.core.db import engine
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
max_tries = 60 * 5 # 5 minutes
wait_seconds = 1
@retry(
stop=stop_after_attempt(max_tries),
wait=wait_fixed(wait_seconds),
before=before_log(logger, logging.INFO),
after=after_log(logger, logging.WARN),
)
def init(db_engine: Engine) -> None:
try:
with Session(db_engine) as session:
# Try to create session to check if DB is awake
session.exec(select(1))
except Exception as e:
logger.error(e)
raise e
def main() -> None:
logger.info("Initializing service")
init(engine)
logger.info("Service finished initializing")
if __name__ == "__main__":
main()
backend/app/core/db.py
-33
View File
@@ -1,33 +0,0 @@
from sqlmodel import Session, create_engine, select
from app import crud
from app.core.config import settings
from app.models import User, UserCreate
engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
# make sure all SQLModel models are imported (app.models) before initializing DB
# otherwise, SQLModel might fail to initialize relationships properly
# for more details: https://github.com/fastapi/full-stack-fastapi-template/issues/28
def init_db(session: Session) -> None:
# Tables should be created with Alembic migrations
# But if you don't want to use migrations, create
# the tables un-commenting the next lines
# from sqlmodel import SQLModel
# This works because the models are already imported and registered from app.models
# SQLModel.metadata.create_all(engine)
user = session.exec(
select(User).where(User.email == settings.FIRST_SUPERUSER)
).first()
if not user:
user_in = UserCreate(
email=settings.FIRST_SUPERUSER,
password=settings.FIRST_SUPERUSER_PASSWORD,
is_superuser=True,
)
user = crud.create_user(session=session, user_create=user_in)
backend/app/crud.py
-54
View File
@@ -1,54 +0,0 @@
import uuid
from typing import Any
from sqlmodel import Session, select
from app.core.security import get_password_hash, verify_password
from app.models import Item, ItemCreate, User, UserCreate, UserUpdate
def create_user(*, session: Session, user_create: UserCreate) -> User:
db_obj = User.model_validate(
user_create, update={"hashed_password": get_password_hash(user_create.password)}
)
session.add(db_obj)
session.commit()
session.refresh(db_obj)
return db_obj
def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
user_data = user_in.model_dump(exclude_unset=True)
extra_data = {}
if "password" in user_data:
password = user_data["password"]
hashed_password = get_password_hash(password)
extra_data["hashed_password"] = hashed_password
db_user.sqlmodel_update(user_data, update=extra_data)
session.add(db_user)
session.commit()
session.refresh(db_user)
return db_user
def get_user_by_email(*, session: Session, email: str) -> User | None:
statement = select(User).where(User.email == email)
session_user = session.exec(statement).first()
return session_user
def authenticate(*, session: Session, email: str, password: str) -> User | None:
db_user = get_user_by_email(session=session, email=email)
if not db_user:
return None
if not verify_password(password, db_user.hashed_password):
return None
return db_user
def create_item(*, session: Session, item_in: ItemCreate, owner_id: uuid.UUID) -> Item:
db_item = Item.model_validate(item_in, update={"owner_id": owner_id})
session.add(db_item)
session.commit()
session.refresh(db_item)
return db_item
backend/app/initial_data.py
-23
View File
@@ -1,23 +0,0 @@
import logging
from sqlmodel import Session
from app.core.db import engine, init_db
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def init() -> None:
with Session(engine) as session:
init_db(session)
def main() -> None:
logger.info("Creating initial data")
init()
logger.info("Initial data created")
if __name__ == "__main__":
main()
backend/app/models.py
-113
View File
@@ -1,113 +0,0 @@
import uuid
from pydantic import EmailStr
from sqlmodel import Field, Relationship, SQLModel
# Shared properties
class UserBase(SQLModel):
email: EmailStr = Field(unique=True, index=True, max_length=255)
is_active: bool = True
is_superuser: bool = False
full_name: str | None = Field(default=None, max_length=255)
# Properties to receive via API on creation
class UserCreate(UserBase):
password: str = Field(min_length=8, max_length=40)
class UserRegister(SQLModel):
email: EmailStr = Field(max_length=255)
password: str = Field(min_length=8, max_length=40)
full_name: str | None = Field(default=None, max_length=255)
# Properties to receive via API on update, all are optional
class UserUpdate(UserBase):
email: EmailStr | None = Field(default=None, max_length=255) # type: ignore
password: str | None = Field(default=None, min_length=8, max_length=40)
class UserUpdateMe(SQLModel):
full_name: str | None = Field(default=None, max_length=255)
email: EmailStr | None = Field(default=None, max_length=255)
class UpdatePassword(SQLModel):
current_password: str = Field(min_length=8, max_length=40)
new_password: str = Field(min_length=8, max_length=40)
# Database model, database table inferred from class name
class User(UserBase, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
hashed_password: str
items: list["Item"] = Relationship(back_populates="owner", cascade_delete=True)
# Properties to return via API, id is always required
class UserPublic(UserBase):
id: uuid.UUID
class UsersPublic(SQLModel):
data: list[UserPublic]
count: int
# Shared properties
class ItemBase(SQLModel):
title: str = Field(min_length=1, max_length=255)
description: str | None = Field(default=None, max_length=255)
# Properties to receive on item creation
class ItemCreate(ItemBase):
pass
# Properties to receive on item update
class ItemUpdate(ItemBase):
title: str | None = Field(default=None, min_length=1, max_length=255) # type: ignore
# Database model, database table inferred from class name
class Item(ItemBase, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
owner_id: uuid.UUID = Field(
foreign_key="user.id", nullable=False, ondelete="CASCADE"
)
owner: User | None = Relationship(back_populates="items")
# Properties to return via API, id is always required
class ItemPublic(ItemBase):
id: uuid.UUID
owner_id: uuid.UUID
class ItemsPublic(SQLModel):
data: list[ItemPublic]
count: int
# Generic message
class Message(SQLModel):
message: str
# JSON payload containing access token
class Token(SQLModel):
access_token: str
token_type: str = "bearer"
# Contents of JWT token
class TokenPayload(SQLModel):
sub: str | None = None
class NewPassword(SQLModel):
token: str
new_password: str = Field(min_length=8, max_length=40)
backend/app/tests/api/routes/test_items.py
-164
View File
@@ -1,164 +0,0 @@
import uuid
from fastapi.testclient import TestClient
from sqlmodel import Session
from app.core.config import settings
from app.tests.utils.item import create_random_item
def test_create_item(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
data = {"title": "Foo", "description": "Fighters"}
response = client.post(
f"{settings.API_V1_STR}/items/",
headers=superuser_token_headers,
json=data,
)
assert response.status_code == 200
content = response.json()
assert content["title"] == data["title"]
assert content["description"] == data["description"]
assert "id" in content
assert "owner_id" in content
def test_read_item(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
item = create_random_item(db)
response = client.get(
f"{settings.API_V1_STR}/items/{item.id}",
headers=superuser_token_headers,
)
assert response.status_code == 200
content = response.json()
assert content["title"] == item.title
assert content["description"] == item.description
assert content["id"] == str(item.id)
assert content["owner_id"] == str(item.owner_id)
def test_read_item_not_found(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
response = client.get(
f"{settings.API_V1_STR}/items/{uuid.uuid4()}",
headers=superuser_token_headers,
)
assert response.status_code == 404
content = response.json()
assert content["detail"] == "Item not found"
def test_read_item_not_enough_permissions(
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
) -> None:
item = create_random_item(db)
response = client.get(
f"{settings.API_V1_STR}/items/{item.id}",
headers=normal_user_token_headers,
)
assert response.status_code == 400
content = response.json()
assert content["detail"] == "Not enough permissions"
def test_read_items(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
create_random_item(db)
create_random_item(db)
response = client.get(
f"{settings.API_V1_STR}/items/",
headers=superuser_token_headers,
)
assert response.status_code == 200
content = response.json()
assert len(content["data"]) >= 2
def test_update_item(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
item = create_random_item(db)
data = {"title": "Updated title", "description": "Updated description"}
response = client.put(
f"{settings.API_V1_STR}/items/{item.id}",
headers=superuser_token_headers,
json=data,
)
assert response.status_code == 200
content = response.json()
assert content["title"] == data["title"]
assert content["description"] == data["description"]
assert content["id"] == str(item.id)
assert content["owner_id"] == str(item.owner_id)
def test_update_item_not_found(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
data = {"title": "Updated title", "description": "Updated description"}
response = client.put(
f"{settings.API_V1_STR}/items/{uuid.uuid4()}",
headers=superuser_token_headers,
json=data,
)
assert response.status_code == 404
content = response.json()
assert content["detail"] == "Item not found"
def test_update_item_not_enough_permissions(
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
) -> None:
item = create_random_item(db)
data = {"title": "Updated title", "description": "Updated description"}
response = client.put(
f"{settings.API_V1_STR}/items/{item.id}",
headers=normal_user_token_headers,
json=data,
)
assert response.status_code == 400
content = response.json()
assert content["detail"] == "Not enough permissions"
def test_delete_item(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
item = create_random_item(db)
response = client.delete(
f"{settings.API_V1_STR}/items/{item.id}",
headers=superuser_token_headers,
)
assert response.status_code == 200
content = response.json()
assert content["message"] == "Item deleted successfully"
def test_delete_item_not_found(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
response = client.delete(
f"{settings.API_V1_STR}/items/{uuid.uuid4()}",
headers=superuser_token_headers,
)
assert response.status_code == 404
content = response.json()
assert content["detail"] == "Item not found"
def test_delete_item_not_enough_permissions(
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
) -> None:
item = create_random_item(db)
response = client.delete(
f"{settings.API_V1_STR}/items/{item.id}",
headers=normal_user_token_headers,
)
assert response.status_code == 400
content = response.json()
assert content["detail"] == "Not enough permissions"
backend/app/tests/api/routes/test_login.py
-118
View File
@@ -1,118 +0,0 @@
from unittest.mock import patch
from fastapi.testclient import TestClient
from sqlmodel import Session
from app.core.config import settings
from app.core.security import verify_password
from app.crud import create_user
from app.models import UserCreate
from app.tests.utils.user import user_authentication_headers
from app.tests.utils.utils import random_email, random_lower_string
from app.utils import generate_password_reset_token
def test_get_access_token(client: TestClient) -> None:
login_data = {
"username": settings.FIRST_SUPERUSER,
"password": settings.FIRST_SUPERUSER_PASSWORD,
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
tokens = r.json()
assert r.status_code == 200
assert "access_token" in tokens
assert tokens["access_token"]
def test_get_access_token_incorrect_password(client: TestClient) -> None:
login_data = {
"username": settings.FIRST_SUPERUSER,
"password": "incorrect",
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
assert r.status_code == 400
def test_use_access_token(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
r = client.post(
f"{settings.API_V1_STR}/login/test-token",
headers=superuser_token_headers,
)
result = r.json()
assert r.status_code == 200
assert "email" in result
def test_recovery_password(
client: TestClient, normal_user_token_headers: dict[str, str]
) -> None:
with (
patch("app.core.config.settings.SMTP_HOST", "smtp.example.com"),
patch("app.core.config.settings.SMTP_USER", "[email protected]"),
):
email = "[email protected]"
r = client.post(
f"{settings.API_V1_STR}/password-recovery/{email}",
headers=normal_user_token_headers,
)
assert r.status_code == 200
assert r.json() == {"message": "Password recovery email sent"}
def test_recovery_password_user_not_exits(
client: TestClient, normal_user_token_headers: dict[str, str]
) -> None:
email = "[email protected]"
r = client.post(
f"{settings.API_V1_STR}/password-recovery/{email}",
headers=normal_user_token_headers,
)
assert r.status_code == 404
def test_reset_password(client: TestClient, db: Session) -> None:
email = random_email()
password = random_lower_string()
new_password = random_lower_string()
user_create = UserCreate(
email=email,
full_name="Test User",
password=password,
is_active=True,
is_superuser=False,
)
user = create_user(session=db, user_create=user_create)
token = generate_password_reset_token(email=email)
headers = user_authentication_headers(client=client, email=email, password=password)
data = {"new_password": new_password, "token": token}
r = client.post(
f"{settings.API_V1_STR}/reset-password/",
headers=headers,
json=data,
)
assert r.status_code == 200
assert r.json() == {"message": "Password updated successfully"}
db.refresh(user)
assert verify_password(new_password, user.hashed_password)
def test_reset_password_invalid_token(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
data = {"new_password": "changethis", "token": "invalid"}
r = client.post(
f"{settings.API_V1_STR}/reset-password/",
headers=superuser_token_headers,
json=data,
)
response = r.json()
assert "detail" in response
assert r.status_code == 400
assert response["detail"] == "Invalid token"
backend/app/tests/api/routes/test_private.py
-26
View File
@@ -1,26 +0,0 @@
from fastapi.testclient import TestClient
from sqlmodel import Session, select
from app.core.config import settings
from app.models import User
def test_create_user(client: TestClient, db: Session) -> None:
r = client.post(
f"{settings.API_V1_STR}/private/users/",
json={
"email": "[email protected]",
"password": "password123",
"full_name": "Pollo Listo",
},
)
assert r.status_code == 200
data = r.json()
user = db.exec(select(User).where(User.id == data["id"])).first()
assert user
assert user.email == "[email protected]"
assert user.full_name == "Pollo Listo"
backend/app/tests/api/routes/test_users.py
-486
View File
@@ -1,486 +0,0 @@
import uuid
from unittest.mock import patch
from fastapi.testclient import TestClient
from sqlmodel import Session, select
from app import crud
from app.core.config import settings
from app.core.security import verify_password
from app.models import User, UserCreate
from app.tests.utils.utils import random_email, random_lower_string
def test_get_users_superuser_me(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
r = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers)
current_user = r.json()
assert current_user
assert current_user["is_active"] is True
assert current_user["is_superuser"]
assert current_user["email"] == settings.FIRST_SUPERUSER
def test_get_users_normal_user_me(
client: TestClient, normal_user_token_headers: dict[str, str]
) -> None:
r = client.get(f"{settings.API_V1_STR}/users/me", headers=normal_user_token_headers)
current_user = r.json()
assert current_user
assert current_user["is_active"] is True
assert current_user["is_superuser"] is False
assert current_user["email"] == settings.EMAIL_TEST_USER
def test_create_user_new_email(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
with (
patch("app.utils.send_email", return_value=None),
patch("app.core.config.settings.SMTP_HOST", "smtp.example.com"),
patch("app.core.config.settings.SMTP_USER", "[email protected]"),
):
username = random_email()
password = random_lower_string()
data = {"email": username, "password": password}
r = client.post(
f"{settings.API_V1_STR}/users/",
headers=superuser_token_headers,
json=data,
)
assert 200 <= r.status_code < 300
created_user = r.json()
user = crud.get_user_by_email(session=db, email=username)
assert user
assert user.email == created_user["email"]
def test_get_existing_user(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
user_id = user.id
r = client.get(
f"{settings.API_V1_STR}/users/{user_id}",
headers=superuser_token_headers,
)
assert 200 <= r.status_code < 300
api_user = r.json()
existing_user = crud.get_user_by_email(session=db, email=username)
assert existing_user
assert existing_user.email == api_user["email"]
def test_get_existing_user_current_user(client: TestClient, db: Session) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
user_id = user.id
login_data = {
"username": username,
"password": password,
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
tokens = r.json()
a_token = tokens["access_token"]
headers = {"Authorization": f"Bearer {a_token}"}
r = client.get(
f"{settings.API_V1_STR}/users/{user_id}",
headers=headers,
)
assert 200 <= r.status_code < 300
api_user = r.json()
existing_user = crud.get_user_by_email(session=db, email=username)
assert existing_user
assert existing_user.email == api_user["email"]
def test_get_existing_user_permissions_error(
client: TestClient, normal_user_token_headers: dict[str, str]
) -> None:
r = client.get(
f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
headers=normal_user_token_headers,
)
assert r.status_code == 403
assert r.json() == {"detail": "The user doesn't have enough privileges"}
def test_create_user_existing_username(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
# username = email
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
crud.create_user(session=db, user_create=user_in)
data = {"email": username, "password": password}
r = client.post(
f"{settings.API_V1_STR}/users/",
headers=superuser_token_headers,
json=data,
)
created_user = r.json()
assert r.status_code == 400
assert "_id" not in created_user
def test_create_user_by_normal_user(
client: TestClient, normal_user_token_headers: dict[str, str]
) -> None:
username = random_email()
password = random_lower_string()
data = {"email": username, "password": password}
r = client.post(
f"{settings.API_V1_STR}/users/",
headers=normal_user_token_headers,
json=data,
)
assert r.status_code == 403
def test_retrieve_users(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
crud.create_user(session=db, user_create=user_in)
username2 = random_email()
password2 = random_lower_string()
user_in2 = UserCreate(email=username2, password=password2)
crud.create_user(session=db, user_create=user_in2)
r = client.get(f"{settings.API_V1_STR}/users/", headers=superuser_token_headers)
all_users = r.json()
assert len(all_users["data"]) > 1
assert "count" in all_users
for item in all_users["data"]:
assert "email" in item
def test_update_user_me(
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
) -> None:
full_name = "Updated Name"
email = random_email()
data = {"full_name": full_name, "email": email}
r = client.patch(
f"{settings.API_V1_STR}/users/me",
headers=normal_user_token_headers,
json=data,
)
assert r.status_code == 200
updated_user = r.json()
assert updated_user["email"] == email
assert updated_user["full_name"] == full_name
user_query = select(User).where(User.email == email)
user_db = db.exec(user_query).first()
assert user_db
assert user_db.email == email
assert user_db.full_name == full_name
def test_update_password_me(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
new_password = random_lower_string()
data = {
"current_password": settings.FIRST_SUPERUSER_PASSWORD,
"new_password": new_password,
}
r = client.patch(
f"{settings.API_V1_STR}/users/me/password",
headers=superuser_token_headers,
json=data,
)
assert r.status_code == 200
updated_user = r.json()
assert updated_user["message"] == "Password updated successfully"
user_query = select(User).where(User.email == settings.FIRST_SUPERUSER)
user_db = db.exec(user_query).first()
assert user_db
assert user_db.email == settings.FIRST_SUPERUSER
assert verify_password(new_password, user_db.hashed_password)
# Revert to the old password to keep consistency in test
old_data = {
"current_password": new_password,
"new_password": settings.FIRST_SUPERUSER_PASSWORD,
}
r = client.patch(
f"{settings.API_V1_STR}/users/me/password",
headers=superuser_token_headers,
json=old_data,
)
db.refresh(user_db)
assert r.status_code == 200
assert verify_password(settings.FIRST_SUPERUSER_PASSWORD, user_db.hashed_password)
def test_update_password_me_incorrect_password(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
new_password = random_lower_string()
data = {"current_password": new_password, "new_password": new_password}
r = client.patch(
f"{settings.API_V1_STR}/users/me/password",
headers=superuser_token_headers,
json=data,
)
assert r.status_code == 400
updated_user = r.json()
assert updated_user["detail"] == "Incorrect password"
def test_update_user_me_email_exists(
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
data = {"email": user.email}
r = client.patch(
f"{settings.API_V1_STR}/users/me",
headers=normal_user_token_headers,
json=data,
)
assert r.status_code == 409
assert r.json()["detail"] == "User with this email already exists"
def test_update_password_me_same_password_error(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
data = {
"current_password": settings.FIRST_SUPERUSER_PASSWORD,
"new_password": settings.FIRST_SUPERUSER_PASSWORD,
}
r = client.patch(
f"{settings.API_V1_STR}/users/me/password",
headers=superuser_token_headers,
json=data,
)
assert r.status_code == 400
updated_user = r.json()
assert (
updated_user["detail"] == "New password cannot be the same as the current one"
)
def test_register_user(client: TestClient, db: Session) -> None:
username = random_email()
password = random_lower_string()
full_name = random_lower_string()
data = {"email": username, "password": password, "full_name": full_name}
r = client.post(
f"{settings.API_V1_STR}/users/signup",
json=data,
)
assert r.status_code == 200
created_user = r.json()
assert created_user["email"] == username
assert created_user["full_name"] == full_name
user_query = select(User).where(User.email == username)
user_db = db.exec(user_query).first()
assert user_db
assert user_db.email == username
assert user_db.full_name == full_name
assert verify_password(password, user_db.hashed_password)
def test_register_user_already_exists_error(client: TestClient) -> None:
password = random_lower_string()
full_name = random_lower_string()
data = {
"email": settings.FIRST_SUPERUSER,
"password": password,
"full_name": full_name,
}
r = client.post(
f"{settings.API_V1_STR}/users/signup",
json=data,
)
assert r.status_code == 400
assert r.json()["detail"] == "The user with this email already exists in the system"
def test_update_user(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
data = {"full_name": "Updated_full_name"}
r = client.patch(
f"{settings.API_V1_STR}/users/{user.id}",
headers=superuser_token_headers,
json=data,
)
assert r.status_code == 200
updated_user = r.json()
assert updated_user["full_name"] == "Updated_full_name"
user_query = select(User).where(User.email == username)
user_db = db.exec(user_query).first()
db.refresh(user_db)
assert user_db
assert user_db.full_name == "Updated_full_name"
def test_update_user_not_exists(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
data = {"full_name": "Updated_full_name"}
r = client.patch(
f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
headers=superuser_token_headers,
json=data,
)
assert r.status_code == 404
assert r.json()["detail"] == "The user with this id does not exist in the system"
def test_update_user_email_exists(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
username2 = random_email()
password2 = random_lower_string()
user_in2 = UserCreate(email=username2, password=password2)
user2 = crud.create_user(session=db, user_create=user_in2)
data = {"email": user2.email}
r = client.patch(
f"{settings.API_V1_STR}/users/{user.id}",
headers=superuser_token_headers,
json=data,
)
assert r.status_code == 409
assert r.json()["detail"] == "User with this email already exists"
def test_delete_user_me(client: TestClient, db: Session) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
user_id = user.id
login_data = {
"username": username,
"password": password,
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
tokens = r.json()
a_token = tokens["access_token"]
headers = {"Authorization": f"Bearer {a_token}"}
r = client.delete(
f"{settings.API_V1_STR}/users/me",
headers=headers,
)
assert r.status_code == 200
deleted_user = r.json()
assert deleted_user["message"] == "User deleted successfully"
result = db.exec(select(User).where(User.id == user_id)).first()
assert result is None
user_query = select(User).where(User.id == user_id)
user_db = db.execute(user_query).first()
assert user_db is None
def test_delete_user_me_as_superuser(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
r = client.delete(
f"{settings.API_V1_STR}/users/me",
headers=superuser_token_headers,
)
assert r.status_code == 403
response = r.json()
assert response["detail"] == "Super users are not allowed to delete themselves"
def test_delete_user_super_user(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
user_id = user.id
r = client.delete(
f"{settings.API_V1_STR}/users/{user_id}",
headers=superuser_token_headers,
)
assert r.status_code == 200
deleted_user = r.json()
assert deleted_user["message"] == "User deleted successfully"
result = db.exec(select(User).where(User.id == user_id)).first()
assert result is None
def test_delete_user_not_found(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
r = client.delete(
f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
headers=superuser_token_headers,
)
assert r.status_code == 404
assert r.json()["detail"] == "User not found"
def test_delete_user_current_super_user_error(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
super_user = crud.get_user_by_email(session=db, email=settings.FIRST_SUPERUSER)
assert super_user
user_id = super_user.id
r = client.delete(
f"{settings.API_V1_STR}/users/{user_id}",
headers=superuser_token_headers,
)
assert r.status_code == 403
assert r.json()["detail"] == "Super users are not allowed to delete themselves"
def test_delete_user_without_privileges(
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
r = client.delete(
f"{settings.API_V1_STR}/users/{user.id}",
headers=normal_user_token_headers,
)
assert r.status_code == 403
assert r.json()["detail"] == "The user doesn't have enough privileges"
backend/app/tests/conftest.py
-42
View File
@@ -1,42 +0,0 @@
from collections.abc import Generator
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, delete
from app.core.config import settings
from app.core.db import engine, init_db
from app.main import app
from app.models import Item, User
from app.tests.utils.user import authentication_token_from_email
from app.tests.utils.utils import get_superuser_token_headers
@pytest.fixture(scope="session", autouse=True)
def db() -> Generator[Session, None, None]:
with Session(engine) as session:
init_db(session)
yield session
statement = delete(Item)
session.execute(statement)
statement = delete(User)
session.execute(statement)
session.commit()
@pytest.fixture(scope="module")
def client() -> Generator[TestClient, None, None]:
with TestClient(app) as c:
yield c
@pytest.fixture(scope="module")
def superuser_token_headers(client: TestClient) -> dict[str, str]:
return get_superuser_token_headers(client)
@pytest.fixture(scope="module")
def normal_user_token_headers(client: TestClient, db: Session) -> dict[str, str]:
return authentication_token_from_email(
client=client, email=settings.EMAIL_TEST_USER, db=db
)
backend/app/tests/crud/test_user.py
-91
View File
@@ -1,91 +0,0 @@
from fastapi.encoders import jsonable_encoder
from sqlmodel import Session
from app import crud
from app.core.security import verify_password
from app.models import User, UserCreate, UserUpdate
from app.tests.utils.utils import random_email, random_lower_string
def test_create_user(db: Session) -> None:
email = random_email()
password = random_lower_string()
user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in)
assert user.email == email
assert hasattr(user, "hashed_password")
def test_authenticate_user(db: Session) -> None:
email = random_email()
password = random_lower_string()
user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in)
authenticated_user = crud.authenticate(session=db, email=email, password=password)
assert authenticated_user
assert user.email == authenticated_user.email
def test_not_authenticate_user(db: Session) -> None:
email = random_email()
password = random_lower_string()
user = crud.authenticate(session=db, email=email, password=password)
assert user is None
def test_check_if_user_is_active(db: Session) -> None:
email = random_email()
password = random_lower_string()
user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in)
assert user.is_active is True
def test_check_if_user_is_active_inactive(db: Session) -> None:
email = random_email()
password = random_lower_string()
user_in = UserCreate(email=email, password=password, disabled=True)
user = crud.create_user(session=db, user_create=user_in)
assert user.is_active
def test_check_if_user_is_superuser(db: Session) -> None:
email = random_email()
password = random_lower_string()
user_in = UserCreate(email=email, password=password, is_superuser=True)
user = crud.create_user(session=db, user_create=user_in)
assert user.is_superuser is True
def test_check_if_user_is_superuser_normal_user(db: Session) -> None:
username = random_email()
password = random_lower_string()
user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in)
assert user.is_superuser is False
def test_get_user(db: Session) -> None:
password = random_lower_string()
username = random_email()
user_in = UserCreate(email=username, password=password, is_superuser=True)
user = crud.create_user(session=db, user_create=user_in)
user_2 = db.get(User, user.id)
assert user_2
assert user.email == user_2.email
assert jsonable_encoder(user) == jsonable_encoder(user_2)
def test_update_user(db: Session) -> None:
password = random_lower_string()
email = random_email()
user_in = UserCreate(email=email, password=password, is_superuser=True)
user = crud.create_user(session=db, user_create=user_in)
new_password = random_lower_string()
user_in_update = UserUpdate(password=new_password, is_superuser=True)
if user.id is not None:
crud.update_user(session=db, db_user=user, user_in=user_in_update)
user_2 = db.get(User, user.id)
assert user_2
assert user.email == user_2.email
assert verify_password(new_password, user_2.hashed_password)
backend/app/tests/scripts/test_backend_pre_start.py
-33
View File
@@ -1,33 +0,0 @@
from unittest.mock import MagicMock, patch
from sqlmodel import select
from app.backend_pre_start import init, logger
def test_init_successful_connection() -> None:
engine_mock = MagicMock()
session_mock = MagicMock()
exec_mock = MagicMock(return_value=True)
session_mock.configure_mock(**{"exec.return_value": exec_mock})
with (
patch("sqlmodel.Session", return_value=session_mock),
patch.object(logger, "info"),
patch.object(logger, "error"),
patch.object(logger, "warn"),
):
try:
init(engine_mock)
connection_successful = True
except Exception:
connection_successful = False
assert (
connection_successful
), "The database connection should be successful and not raise an exception."
assert session_mock.exec.called_once_with(
select(1)
), "The session should execute a select statement once."
backend/app/tests/scripts/test_test_pre_start.py
-33
View File
@@ -1,33 +0,0 @@
from unittest.mock import MagicMock, patch
from sqlmodel import select
from app.tests_pre_start import init, logger
def test_init_successful_connection() -> None:
engine_mock = MagicMock()
session_mock = MagicMock()
exec_mock = MagicMock(return_value=True)
session_mock.configure_mock(**{"exec.return_value": exec_mock})
with (
patch("sqlmodel.Session", return_value=session_mock),
patch.object(logger, "info"),
patch.object(logger, "error"),
patch.object(logger, "warn"),
):
try:
init(engine_mock)
connection_successful = True
except Exception:
connection_successful = False
assert (
connection_successful
), "The database connection should be successful and not raise an exception."
assert session_mock.exec.called_once_with(
select(1)
), "The session should execute a select statement once."
backend/app/tests/utils/item.py
-16
View File
@@ -1,16 +0,0 @@
from sqlmodel import Session
from app import crud
from app.models import Item, ItemCreate
from app.tests.utils.user import create_random_user
from app.tests.utils.utils import random_lower_string
def create_random_item(db: Session) -> Item:
user = create_random_user(db)
owner_id = user.id
assert owner_id is not None
title = random_lower_string()
description = random_lower_string()
item_in = ItemCreate(title=title, description=description)
return crud.create_item(session=db, item_in=item_in, owner_id=owner_id)
backend/app/tests/utils/user.py
-49
View File
@@ -1,49 +0,0 @@
from fastapi.testclient import TestClient
from sqlmodel import Session
from app import crud
from app.core.config import settings
from app.models import User, UserCreate, UserUpdate
from app.tests.utils.utils import random_email, random_lower_string
def user_authentication_headers(
*, client: TestClient, email: str, password: str
) -> dict[str, str]:
data = {"username": email, "password": password}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=data)
response = r.json()
auth_token = response["access_token"]
headers = {"Authorization": f"Bearer {auth_token}"}
return headers
def create_random_user(db: Session) -> User:
email = random_email()
password = random_lower_string()
user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in)
return user
def authentication_token_from_email(
*, client: TestClient, email: str, db: Session
) -> dict[str, str]:
"""
Return a valid token for the user with given email.
If the user doesn't exist it is created first.
"""
password = random_lower_string()
user = crud.get_user_by_email(session=db, email=email)
if not user:
user_in_create = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in_create)
else:
user_in_update = UserUpdate(password=password)
if not user.id:
raise Exception("User id not set")
user = crud.update_user(session=db, db_user=user, user_in=user_in_update)
return user_authentication_headers(client=client, email=email, password=password)
backend/app/tests/utils/utils.py
-26
View File
@@ -1,26 +0,0 @@
import random
import string
from fastapi.testclient import TestClient
from app.core.config import settings
def random_lower_string() -> str:
return "".join(random.choices(string.ascii_lowercase, k=32))
def random_email() -> str:
return f"{random_lower_string()}@{random_lower_string()}.com"
def get_superuser_token_headers(client: TestClient) -> dict[str, str]:
login_data = {
"username": settings.FIRST_SUPERUSER,
"password": settings.FIRST_SUPERUSER_PASSWORD,
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
tokens = r.json()
a_token = tokens["access_token"]
headers = {"Authorization": f"Bearer {a_token}"}
return headers
backend/app/tests_pre_start.py
-39
View File
@@ -1,39 +0,0 @@
import logging
from sqlalchemy import Engine
from sqlmodel import Session, select
from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed
from app.core.db import engine
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
max_tries = 60 * 5 # 5 minutes
wait_seconds = 1
@retry(
stop=stop_after_attempt(max_tries),
wait=wait_fixed(wait_seconds),
before=before_log(logger, logging.INFO),
after=after_log(logger, logging.WARN),
)
def init(db_engine: Engine) -> None:
try:
# Try to create session to check if DB is awake
with Session(db_engine) as session:
session.exec(select(1))
except Exception as e:
logger.error(e)
raise e
def main() -> None:
logger.info("Initializing service")
init(engine)
logger.info("Service finished initializing")
if __name__ == "__main__":
main()
backend/app/__init__.py → backend/apps/__init__.py
View File
backend/apps/api.py
+10
View File
@@ -0,0 +1,10 @@
from fastapi import APIRouter
from apps.system.api import login, user
api_router = APIRouter()
api_router.include_router(login.router)
api_router.include_router(user.router)
backend/app/api/__init__.py → backend/apps/system/__init__.py
View File
backend/app/api/routes/__init__.py → backend/apps/system/api/__init__.py
View File
backend/apps/system/api/login.py
+22
View File
@@ -0,0 +1,22 @@
from fastapi import APIRouter, HTTPException
from apps.system.schemas.auth import LocalLoginSchema
from common.core.deps import SessionDep
from ..crud.user import authenticate
from common.core.security import create_access_token
from datetime import timedelta
from common.core.config import settings
router = APIRouter(tags=["login"])
@router.post("/localLogin")
def local_login(
session: SessionDep,
schema: LocalLoginSchema,
) -> str:
user = authenticate(session=session, account=schema.account, password=schema.password)
if not user:
raise HTTPException(status_code=400, detail="Incorrect account or password")
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return create_access_token(
user.id, expires_delta=access_token_expires
)
backend/apps/system/api/user.py
+11
View File
@@ -0,0 +1,11 @@
from fastapi import APIRouter, Depends, Request
from common.core.deps import CurrentUser
router = APIRouter(tags=["user"], prefix="/user")
@router.get("/info")
async def user_info():
return {
"name": "admin"
}
backend/app/core/__init__.py → backend/apps/system/crud/__init__.py
View File
backend/apps/system/crud/user.py
+17
View File
@@ -0,0 +1,17 @@
from sqlmodel import Session, select
from ..models.user import sys_user
from common.core.security import verify_md5pwd
def get_user_by_account(*, session: Session, account: str) -> sys_user | None:
statement = select(sys_user).where(sys_user.account == account)
session_user = session.exec(statement).first()
return session_user
def authenticate(*, session: Session, account: str, password: str) -> sys_user | None:
db_user = get_user_by_account(session=session, account=account)
if not db_user:
return None
if not verify_md5pwd(password, db_user.password):
return None
return db_user
backend/app/tests/__init__.py → backend/apps/system/middleware/__init__.py
View File
backend/apps/system/middleware/token.py
+23
View File
@@ -0,0 +1,23 @@
from fastapi import HTTPException, Request
from common.core.config import settings
from common.core.deps import get_current_user
async def verify_token(request: Request, call_next):
print(request.url.path)
if request.url.path == "/login/localLogin":
return await call_next(request)
try:
key = settings.TOKEN_KEY
token = request.headers.get(key)
get_current_user(token)
except (Exception):
return HTTPException(status_code=401, detail="Invalid token")
response = await call_next(request)
return response
backend/app/tests/api/__init__.py → backend/apps/system/models/__init__.py
View File
backend/apps/system/models/user.py
+8
View File
@@ -0,0 +1,8 @@
from sqlmodel import SQLModel, Field
class sys_user(SQLModel, table=True):
id: int = Field(primary_key=True, index=True)
account: str = Field(max_length=255, unique=True)
password: str = Field(max_length=255)
oid: int = Field(default=1)
backend/app/tests/api/routes/__init__.py → backend/apps/system/schemas/__init__.py
View File
backend/apps/system/schemas/auth.py
+7
View File
@@ -0,0 +1,7 @@
from pydantic import BaseModel
class LocalLoginSchema(BaseModel):
account: str
password: str
backend/app/tests/crud/__init__.py → backend/common/__init__.py
View File
backend/app/tests/scripts/__init__.py → backend/common/core/__init__.py
View File
backend/app/core/config.py → backend/common/core/config.py
+2
View File
@@ -56,6 +56,8 @@ class Settings(BaseSettings):
POSTGRES_USER: str
POSTGRES_PASSWORD: str = ""
POSTGRES_DB: str = ""
TOKEN_KEY: str
@computed_field # type: ignore[prop-decorator]
@property
backend/common/core/db.py
+14
View File
@@ -0,0 +1,14 @@
from sqlmodel import Session, create_engine, SQLModel
from common.core.config import settings
engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
def get_session():
with Session(engine) as session:
yield session
def init_db():
SQLModel.metadata.create_all(engine)
backend/common/core/deps.py
+73
View File
@@ -0,0 +1,73 @@
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pydantic import ValidationError
from sqlmodel import Session
from common.core import security
from common.core.config import settings
from common.core.db import get_session
from apps.system.models.user import sys_user
import threading
""" reusable_oauth2 = OAuth2PasswordBearer(
tokenUrl=f"{settings.API_V1_STR}/login/access-token"
)
"""
thread_local = threading.local()
SessionDep = Annotated[Session, Depends(get_session)]
# TokenDep = Annotated[str, Depends(reusable_oauth2)]
def get_current_user(session: SessionDep, token: str) -> sys_user:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
)
# token_data = TokenPayload(**payload)
except (InvalidTokenError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = session.get(sys_user, payload.sub)
if not user:
raise HTTPException(status_code=404, detail="User not found")
""" if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user") """
thread_local.current_user = user
return user
def get_current_user_from_thread():
return getattr(thread_local, "current_user", None)
CurrentUser = Annotated[sys_user, Depends(get_current_user_from_thread)]
""" def get_current_user(session: SessionDep, token: TokenDep) -> User:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
)
token_data = TokenPayload(**payload)
except (InvalidTokenError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = session.get(User, token_data.sub)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return user
CurrentUser = Annotated[User, Depends(get_current_user)]
def get_current_active_superuser(current_user: CurrentUser) -> User:
if not current_user.is_superuser:
raise HTTPException(
status_code=403, detail="The user doesn't have enough privileges"
)
return current_user """
backend/app/core/security.py → backend/common/core/security.py
+11 -3
View File
@@ -3,8 +3,8 @@ from typing import Any
import jwt
from passlib.context import CryptContext
from app.core.config import settings
import hashlib
from common.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
@@ -14,7 +14,7 @@ ALGORITHM = "HS256"
def create_access_token(subject: str | Any, expires_delta: timedelta) -> str:
expire = datetime.now(timezone.utc) + expires_delta
to_encode = {"exp": expire, "sub": str(subject)}
to_encode = {"exp": expire, "account": str(subject)}
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
@@ -25,3 +25,11 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def md5pwd(password: str) -> str:
m = hashlib.md5()
m.update(password.encode("utf-8"))
return m.hexdigest()
def verify_md5pwd(plain_password: str, md5_password: str) -> bool:
return md5pwd(plain_password) == md5_password
backend/app/tests/utils/__init__.py → backend/common/utils/__init__.py
View File
backend/app/utils.py → backend/common/utils/utils.py
+3 -3
View File
@@ -9,10 +9,10 @@ import jwt
from jwt.exceptions import InvalidTokenError
from app.core import security
from app.core.config import settings
from common.core import security
from common.core.config import settings
logging.basicConfig(level=logging.INFO)
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
backend/app/main.py → backend/main.py
+9 -3
View File
@@ -2,10 +2,11 @@ import sentry_sdk
from fastapi import FastAPI
from fastapi.routing import APIRoute
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from apps.api import api_router
from common.core.config import settings
from app.api.main import api_router
from app.core.config import settings
from apps.system.middleware.token import verify_token
def custom_generate_unique_id(route: APIRoute) -> str:
return f"{route.tags[0]}-{route.name}"
@@ -31,3 +32,8 @@ if settings.all_cors_origins:
)
app.include_router(api_router, prefix=settings.API_V1_STR)
# app.add_middleware(BaseHTTPMiddleware, dispatch=verify_token)
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)