Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file modified .coverage
Binary file not shown.
20 changes: 18 additions & 2 deletions .github/workflows/create_pull_request.yml
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ jobs:
python -m pip install --upgrade pip
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
pip install pytest # Ensures pytest is available even if not in requirements.txt
pip install pytest pytest-cov

- name: Create .env file
run: |
Expand All @@ -55,11 +56,26 @@ jobs:
echo "MAIL_PORT=465" >> .env
echo "MAIL_SERVER=smtp.gmail.com" >> .env


- name: Run Tests
id: coverage_step
run: |
mkdir logs
pytest -s
coverage run -m pytest
PERCENT=$(coverage report | grep TOTAL | awk '{print $NF}' | sed 's/%//')
echo "PERCENTAGE=$PERCENT" >> $GITHUB_OUTPUT

- name: Create Coverage Badge
uses: schneegans/dynamic-badges-action@v1.7.0
with:
auth: ${{ secrets.GIST_SECRET }}
gistID: b56b3d61a5e739fd26252cda094bace2
filename: fastapi_project_structure_coverage.json
label: Coverage
message: ${{ steps.coverage_step.outputs.PERCENTAGE }}%
valColorRange: ${{ steps.coverage_step.outputs.PERCENTAGE }}
maxColorRange: 100
minColorRange: 0


create_pull_request:
runs-on: ubuntu-latest
Expand Down
3 changes: 3 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
# FastAPI Project Structure Template
⚙️ You can Generate Project Interactively Based on this template with the [FastAPI Gen8 CLI Tool](https://pypi.org/project/fastapi-gen8/)


![Coverage Badge](https://img.shields.io/endpoint?url=https://gist.githubusercontent.com/brianobot/b56b3d61a5e739fd26252cda094bace2/raw/fastapi_project_structure_coverage.json)

[📖 Read Article here](https://medium.com/@brianobot9/the-ultimate-fastapi-project-blueprint-build-scalable-secure-and-maintainable-systems-with-ease-acbc4e058012)

This repository provides a clean and scalable template for building FastAPI applications. It is designed to help you start new projects quickly with best practices in mind.
Expand Down
12 changes: 8 additions & 4 deletions app/services/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,7 @@ async def reset_password(

user = await get_user(reset_data.email, session)
if not user:
raise HTTPException(status_code=404, detail="User not found")
raise HTTPException(status_code=404, detail="Invalid Reset Code")

hashed_password = get_password_hash(reset_data.new_password)
stmt = (
Expand Down Expand Up @@ -201,7 +201,7 @@ async def update_user(
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
detail="Incorrect Old Password",
headers={"WWW-Authenticate": "Bearer"},
)
data.update({"password": get_password_hash(new_password)})
Expand All @@ -218,7 +218,9 @@ async def update_user(
return result.scalar_one()


def create_access_token(data: dict, expires_delta: timedelta | None = None):
def create_access_token(
data: dict[str, str | datetime], expires_delta: timedelta | None = None
):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(UTC) + expires_delta
Expand All @@ -229,7 +231,9 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None):
return encoded_jwt


def create_refresh_token(data: dict, expires_delta: timedelta | None = None):
def create_refresh_token(
data: dict[str, str | datetime], expires_delta: timedelta | None = None
):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(UTC) + expires_delta
Expand Down
87 changes: 80 additions & 7 deletions app/services/tests/test_auth.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from datetime import timedelta
from typing import Any

import pytest
Expand Down Expand Up @@ -109,11 +110,9 @@ async def test_initiate_password_reset_for_user(user: UserDB, session: AsyncSess
assert result == {"detail": "Password reset code sent"}


async def test_initiate_password_reset_for_none_user(
user: UserDB, session: AsyncSession
):
async def test_initiate_password_reset_for_none_user(session: AsyncSession):
result = await auth_services.initiate_password_reset(
user.email, session, BackgroundTasks(tasks=[])
faker.email(), session, BackgroundTasks(tasks=[])
)
assert result == {"detail": "Password reset code sent"}

Expand All @@ -133,14 +132,19 @@ async def test_verify_reset_password_otp_for_user(user: UserDB, session: AsyncSe
],
)
async def test_verify_reset_password_otp_fails(session: AsyncSession, code: str):
redis_manager.cache_json_item(f"reset-code-{faker.email()}", {"code": "000000"})
none_existence_email = faker.email()
redis_manager.cache_json_item(
f"reset-code-{none_existence_email}", {"code": "000000"}
)
with pytest.raises(HTTPException) as err:
await auth_services.verify_reset_password_otp(code, faker.email(), session)
await auth_services.verify_reset_password_otp(
code, none_existence_email, session
)

assert err.value.detail == "Invalid Reset Code"


async def test_reset_password(user: UserDB, session: AsyncSession):
async def test_reset_password_for_user(user: UserDB, session: AsyncSession):
redis_manager.cache_json_item(f"reset-code-{user.email}", {"code": "000000"})
result = await auth_services.reset_password(
auth_schemas.PasswordResetData(
Expand All @@ -149,3 +153,72 @@ async def test_reset_password(user: UserDB, session: AsyncSession):
session,
)
assert result == {"detail": "Password reset successfully"}


@pytest.mark.parametrize(
"code",
[
"000000",
"000001",
],
)
async def test_reset_password_fails(user: UserDB, session: AsyncSession, code: str):
none_existence_email = faker.email()
redis_manager.cache_json_item(
f"reset-code-{none_existence_email}", {"code": "000000"}
)
with pytest.raises(HTTPException) as err:
await auth_services.reset_password(
auth_schemas.PasswordResetData(
code=code, email=none_existence_email, new_password="newpassword"
),
session,
)

assert err.value.detail == "Invalid Reset Code"


async def test_update_user(user: UserDB, session: AsyncSession):
updated_user = await auth_services.update_user(
user.email,
auth_schemas.UpdateUserModel(
old_password="password", new_password="new_password"
),
session,
)
assert isinstance(updated_user, UserDB)
assert auth_services.verify_password("new_password", updated_user.password)


async def test_update_user_fails(user: UserDB, session: AsyncSession):
with pytest.raises(HTTPException) as err:
await auth_services.update_user(
user.email,
auth_schemas.UpdateUserModel(
old_password="incorrectpassword", new_password="new_password"
),
session,
)
assert err.value.detail == "Incorrect Old Password"


@pytest.mark.parametrize(
"expires_delta",
[
None,
timedelta(days=10),
],
)
async def test_create_access_token(expires_delta: timedelta | None):
auth_services.create_access_token({"sub": faker.email()}, expires_delta)


@pytest.mark.parametrize(
"expires_delta",
[
None,
timedelta(days=10),
],
)
async def test_refresh_access_token(expires_delta: timedelta | None):
auth_services.create_refresh_token({"sub": faker.email()}, expires_delta)