mirror of
https://github.com/furyhawk/listen.git
synced 2026-07-20 08:25:34 +00:00
Initial commit
This commit is contained in:
@@ -0,0 +1,9 @@
|
||||
SECURITY__JWT_SECRET_KEY=DVnFmhwvjEhJZpuhndxjhlezxQPJmBIIkMDEmFREWQADPcUnrG
|
||||
SECURITY__BACKEND_CORS_ORIGINS=["http://localhost:3000","http://localhost:8001"]
|
||||
SECURITY__ALLOWED_HOSTS=["localhost", "127.0.0.1"]
|
||||
|
||||
DATABASE__HOSTNAME=localhost
|
||||
DATABASE__USERNAME=rDGJeEDqAz
|
||||
DATABASE__PASSWORD=XsPQhCoEfOQZueDjsILetLDUvbvSxAMnrVtgVZpmdcSssUgbvs
|
||||
DATABASE__PORT=5455
|
||||
DATABASE__DB=default_db
|
||||
@@ -0,0 +1,25 @@
|
||||
version: 2
|
||||
updates:
|
||||
- package-ecosystem: pip
|
||||
directory: /
|
||||
schedule:
|
||||
interval: weekly
|
||||
open-pull-requests-limit: 5
|
||||
allow:
|
||||
- dependency-type: "all"
|
||||
groups:
|
||||
all-dependencies:
|
||||
patterns:
|
||||
- "*"
|
||||
exclude-patterns:
|
||||
- "pytest-asyncio"
|
||||
|
||||
- package-ecosystem: github-actions
|
||||
directory: /
|
||||
schedule:
|
||||
interval: weekly
|
||||
|
||||
- package-ecosystem: docker
|
||||
directory: /
|
||||
schedule:
|
||||
interval: weekly
|
||||
@@ -0,0 +1,36 @@
|
||||
name: dev-build
|
||||
on:
|
||||
workflow_run:
|
||||
workflows: ["tests"]
|
||||
branches: [main]
|
||||
types:
|
||||
- completed
|
||||
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
tag:
|
||||
description: "Docker image tag"
|
||||
required: true
|
||||
default: "latest"
|
||||
|
||||
env:
|
||||
IMAGE_TAG: ${{ github.event.inputs.tag || 'latest' }}
|
||||
|
||||
jobs:
|
||||
dev_build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Login to DockerHub
|
||||
uses: docker/login-action@v3
|
||||
with:
|
||||
username: ${{ secrets.DOCKER_USER }}
|
||||
password: ${{ secrets.DOCKER_PASS }}
|
||||
|
||||
- name: Build and push image
|
||||
uses: docker/build-push-action@v5
|
||||
with:
|
||||
file: Dockerfile
|
||||
push: true
|
||||
tags: rafsaf/minimal-fastapi-postgres-template:${{ env.IMAGE_TAG }}
|
||||
@@ -0,0 +1,57 @@
|
||||
name: tests
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "**"
|
||||
tags-ignore:
|
||||
- "*.*"
|
||||
|
||||
jobs:
|
||||
tests:
|
||||
runs-on: ubuntu-latest
|
||||
services:
|
||||
postgres:
|
||||
image: postgres
|
||||
env:
|
||||
POSTGRES_PASSWORD: postgres
|
||||
options: >-
|
||||
--health-cmd pg_isready
|
||||
--health-interval 10s
|
||||
--health-timeout 5s
|
||||
--health-retries 5
|
||||
ports:
|
||||
- 5432:5432
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12.2"
|
||||
|
||||
- name: Install Poetry
|
||||
uses: snok/install-poetry@v1
|
||||
with:
|
||||
virtualenvs-create: true
|
||||
virtualenvs-in-project: false
|
||||
virtualenvs-path: /opt/venv
|
||||
|
||||
- name: Load cached venv
|
||||
id: cached-poetry-dependencies
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: /opt/venv
|
||||
key: venv-${{ runner.os }}-python-3.12.2-${{ hashFiles('poetry.lock') }}
|
||||
|
||||
- name: Install dependencies and actiavte virtualenv
|
||||
if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
|
||||
run: |
|
||||
poetry install --no-interaction --no-root
|
||||
|
||||
- name: Run tests
|
||||
env:
|
||||
SECURITY__JWT_SECRET_KEY: very-not-secret
|
||||
DATABASE__HOSTNAME: localhost
|
||||
DATABASE__PASSWORD: postgres
|
||||
run: |
|
||||
poetry run pytest
|
||||
@@ -0,0 +1,51 @@
|
||||
name: type-check
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "**"
|
||||
tags-ignore:
|
||||
- "*.*"
|
||||
|
||||
jobs:
|
||||
type_check:
|
||||
strategy:
|
||||
matrix:
|
||||
check: ["ruff check", "mypy --check", "ruff format --check"]
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12.2"
|
||||
|
||||
- name: Install Poetry
|
||||
uses: snok/install-poetry@v1
|
||||
with:
|
||||
virtualenvs-create: true
|
||||
virtualenvs-in-project: true
|
||||
|
||||
- name: Install Poetry
|
||||
uses: snok/install-poetry@v1
|
||||
with:
|
||||
virtualenvs-create: true
|
||||
virtualenvs-in-project: false
|
||||
virtualenvs-path: /opt/venv
|
||||
|
||||
- name: Load cached venv
|
||||
id: cached-poetry-dependencies
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: /opt/venv
|
||||
key: venv-${{ runner.os }}-python-3.12.2-${{ hashFiles('poetry.lock') }}
|
||||
|
||||
- name: Install dependencies and actiavte virtualenv
|
||||
if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
|
||||
run: |
|
||||
poetry install --no-interaction --no-root
|
||||
|
||||
- name: Run ${{ matrix.check }}
|
||||
run: |
|
||||
poetry run ${{ matrix.check }} .
|
||||
+134
@@ -0,0 +1,134 @@
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
.env
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
pip-wheel-metadata/
|
||||
share/python-wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
log.txt
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
local_settings.py
|
||||
db.sqlite3
|
||||
db.sqlite3-journal
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# IPython
|
||||
profile_default/
|
||||
ipython_config.py
|
||||
|
||||
# pyenv
|
||||
.python-version
|
||||
|
||||
# pipenv
|
||||
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
|
||||
# However, in case of collaboration, if having platform-specific dependencies or dependencies
|
||||
# having no cross-platform support, pipenv may install dependencies that don't work, or not
|
||||
# install all needed dependencies.
|
||||
#Pipfile.lock
|
||||
|
||||
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
|
||||
__pypackages__/
|
||||
|
||||
# Celery stuff
|
||||
celerybeat-schedule
|
||||
celerybeat.pid
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
.dmypy.json
|
||||
dmypy.json
|
||||
|
||||
# ruff
|
||||
.ruff_cache
|
||||
|
||||
# Pyre type checker
|
||||
.pyre/
|
||||
@@ -0,0 +1,16 @@
|
||||
repos:
|
||||
- repo: https://github.com/pre-commit/pre-commit-hooks
|
||||
rev: v4.5.0
|
||||
hooks:
|
||||
- id: check-yaml
|
||||
|
||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||
rev: v0.3.0
|
||||
hooks:
|
||||
- id: ruff-format
|
||||
|
||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||
rev: v0.3.0
|
||||
hooks:
|
||||
- id: ruff
|
||||
args: [--fix]
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
FROM python:3.12.2-slim-bullseye as base
|
||||
|
||||
ENV PYTHONUNBUFFERED 1
|
||||
WORKDIR /build
|
||||
|
||||
# Create requirements.txt file
|
||||
FROM base as poetry
|
||||
RUN pip install poetry==1.8.2
|
||||
COPY poetry.lock pyproject.toml ./
|
||||
RUN poetry export -o /requirements.txt --without-hashes
|
||||
|
||||
FROM base as common
|
||||
COPY --from=poetry /requirements.txt .
|
||||
# Create venv, add it to path and install requirements
|
||||
RUN python -m venv /venv
|
||||
ENV PATH="/venv/bin:$PATH"
|
||||
RUN pip install -r requirements.txt
|
||||
|
||||
# Install uvicorn server
|
||||
RUN pip install uvicorn[standard]
|
||||
|
||||
# Copy the rest of app
|
||||
COPY app app
|
||||
COPY alembic alembic
|
||||
COPY alembic.ini .
|
||||
COPY pyproject.toml .
|
||||
COPY init.sh .
|
||||
|
||||
# Create new user to run app process as unprivilaged user
|
||||
RUN addgroup --gid 1001 --system uvicorn && \
|
||||
adduser --gid 1001 --shell /bin/false --disabled-password --uid 1001 uvicorn
|
||||
|
||||
# Run init.sh script then start uvicorn
|
||||
RUN chown -R uvicorn:uvicorn /build
|
||||
CMD bash init.sh && \
|
||||
runuser -u uvicorn -- /venv/bin/uvicorn app.main:app --app-dir /build --host 0.0.0.0 --port 8000 --workers 2 --loop uvloop
|
||||
EXPOSE 8000
|
||||
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2021 rafsaf
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -0,0 +1,410 @@
|
||||
[](https://minimal-fastapi-postgres-template.rafsaf.pl/)
|
||||
[](https://github.com/rafsaf/minimal-fastapi-postgres-template/blob/main/LICENSE)
|
||||
[](https://docs.python.org/3/whatsnew/3.12.html)
|
||||
[](https://github.com/astral-sh/ruff)
|
||||
[](https://github.com/rafsaf/minimal-fastapi-postgres-template/actions/workflows/tests.yml)
|
||||
|
||||
_Check out online example: https://minimal-fastapi-postgres-template.rafsaf.pl, it's 100% code used in template (docker image) with added my domain and https only._
|
||||
|
||||
# Minimal async FastAPI + PostgreSQL template
|
||||
|
||||
- [Minimal async FastAPI + PostgreSQL template](#minimal-async-fastapi--postgresql-template)
|
||||
- [Features](#features)
|
||||
- [Quickstart](#quickstart)
|
||||
- [1. Create repository from a template](#1-create-repository-from-a-template)
|
||||
- [2. Install dependecies with Poetry](#2-install-dependecies-with-poetry)
|
||||
- [3. Setup database and migrations](#3-setup-database-and-migrations)
|
||||
- [4. Now you can run app](#4-now-you-can-run-app)
|
||||
- [5. Activate pre-commit](#5-activate-pre-commit)
|
||||
- [6. Running tests](#6-running-tests)
|
||||
- [About](#about)
|
||||
- [Step by step example - POST and GET endpoints](#step-by-step-example---post-and-get-endpoints)
|
||||
- [1. Create SQLAlchemy model](#1-create-sqlalchemy-model)
|
||||
- [2. Create and apply alembic migration](#2-create-and-apply-alembic-migration)
|
||||
- [3. Create request and response schemas](#3-create-request-and-response-schemas)
|
||||
- [4. Create endpoints](#4-create-endpoints)
|
||||
- [5. Write tests](#5-write-tests)
|
||||
- [Design](#design)
|
||||
- [Deployment strategies - via Docker image](#deployment-strategies---via-docker-image)
|
||||
- [Docs URL, CORS and Allowed Hosts](#docs-url-cors-and-allowed-hosts)
|
||||
- [License](#license)
|
||||
|
||||
|
||||
## Features
|
||||
|
||||
- [x] Template repository
|
||||
- [x] SQLAlchemy 2.0, async queries, best possible autocompletion support
|
||||
- [x] PostgreSQL 16 database under `asyncpg`, docker-compose.yml
|
||||
- [x] Full [Alembic](https://alembic.sqlalchemy.org/en/latest/) migrations setup
|
||||
- [x] Refresh token endpoint (not only access like in official template)
|
||||
- [x] Ready to go Dockerfile with [uvicorn](https://www.uvicorn.org/) webserver as an example
|
||||
- [x] [Poetry](https://python-poetry.org/docs/), `mypy`, `pre-commit` hooks with [ruff](https://github.com/astral-sh/ruff)
|
||||
- [x] Perfect pytest asynchronous test setup with +40 tests and full coverage
|
||||
|
||||
<br>
|
||||
|
||||
|
||||
|
||||
<kbd></kbd>
|
||||
|
||||
|
||||
|
||||
## Quickstart
|
||||
|
||||
### 1. Create repository from a template
|
||||
|
||||
See [docs](https://docs.github.com/en/repositories/creating-and-managing-repositories/creating-a-repository-from-a-template).
|
||||
|
||||
### 2. Install dependecies with [Poetry](https://python-poetry.org/docs/)
|
||||
|
||||
```bash
|
||||
cd your_project_name
|
||||
|
||||
### Poetry install (python3.12)
|
||||
poetry install
|
||||
```
|
||||
|
||||
Note, be sure to use `python3.12` with this template with either poetry or standard venv & pip, if you need to stick to some earlier python version, you should adapt it yourself (remove new versions specific syntax for example `str | int` for python < 3.10)
|
||||
|
||||
### 3. Setup database and migrations
|
||||
|
||||
```bash
|
||||
### Setup database
|
||||
docker-compose up -d
|
||||
|
||||
### Run Alembic migrations
|
||||
alembic upgrade head
|
||||
```
|
||||
|
||||
### 4. Now you can run app
|
||||
|
||||
```bash
|
||||
### And this is it:
|
||||
uvicorn app.main:app --reload
|
||||
|
||||
```
|
||||
|
||||
You should then use `git init` (if needed) to initialize git repository and access OpenAPI spec at http://localhost:8000/ by default. To customize docs url, cors and allowed hosts settings, read [section about it](#docs-url-cors-and-allowed-hosts).
|
||||
|
||||
### 5. Activate pre-commit
|
||||
|
||||
[pre-commit](https://pre-commit.com/) is de facto standard now for pre push activities like isort or black or its nowadays replacement ruff.
|
||||
|
||||
Refer to `.pre-commit-config.yaml` file to see my current opinionated choices.
|
||||
|
||||
```bash
|
||||
# Install pre-commit
|
||||
pre-commit install --install-hooks
|
||||
|
||||
# Run on all files
|
||||
pre-commit run --all-files
|
||||
```
|
||||
|
||||
### 6. Running tests
|
||||
|
||||
Note, it will create databases for session and run tests in many processes by default (using pytest-xdist) to speed up execution, based on how many CPU are available in environment.
|
||||
|
||||
For more details about initial database setup, see logic `app/tests/conftest.py` file, `fixture_setup_new_test_database` function.
|
||||
|
||||
Moreover, there is coverage pytest plugin with required code coverage level 100%.
|
||||
|
||||
```bash
|
||||
# see all pytest configuration flags in pyproject.toml
|
||||
pytest
|
||||
```
|
||||
|
||||
<br>
|
||||
|
||||
## About
|
||||
|
||||
This project is heavily based on the official template https://github.com/tiangolo/full-stack-fastapi-postgresql (and on my previous work: [link1](https://github.com/rafsaf/fastapi-plan), [link2](https://github.com/rafsaf/docker-fastapi-projects)), but as it now not too much up-to-date, it is much easier to create new one than change official. I didn't like some of conventions over there also (`crud` and `db` folders for example or `schemas` with bunch of files). This template aims to be as much up-to-date as possible, using only newest python versions and libraries versions.
|
||||
|
||||
`2.0` style SQLAlchemy API is good enough so there is no need to write everything in `crud` and waste our time... The `core` folder was also rewritten. There is great base for writting tests in `tests`, but I didn't want to write hundreds of them, I noticed that usually after changes in the structure of the project, auto tests are useless and you have to write them from scratch anyway (delete old ones...), hence less than more. Similarly with the `User` model, it is very modest, with just `id` (uuid), `email` and `password_hash`, because it will be adapted to the project anyway.
|
||||
|
||||
2024 update:
|
||||
|
||||
The template was adpoted to my current style and knowledge, the test based expanded to cover more, added mypy, ruff and test setup was completly rewritten to have three things:
|
||||
|
||||
- run test in paraller in many processes for speed
|
||||
- transactions rollback after every test
|
||||
- create test databases instead of having another in docker-compose.yml
|
||||
|
||||
<br>
|
||||
|
||||
## Step by step example - POST and GET endpoints
|
||||
|
||||
I always enjoy to have some kind of an example in templates (even if I don't like it much, _some_ parts may be useful and save my time...), so let's create two example endpoints:
|
||||
|
||||
- `POST` endpoint `/pets/create` for creating `Pets` with relation to currently logged `User`
|
||||
- `GET` endpoint `/pets/me` for fetching all user's pets.
|
||||
|
||||
<br>
|
||||
|
||||
### 1. Create SQLAlchemy model
|
||||
|
||||
We will add `Pet` model to `app/models.py`.
|
||||
|
||||
```python
|
||||
# app/models.py
|
||||
|
||||
(...)
|
||||
|
||||
class Pet(Base):
|
||||
__tablename__ = "pet"
|
||||
|
||||
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
|
||||
user_id: Mapped[str] = mapped_column(
|
||||
ForeignKey("user_account.user_id", ondelete="CASCADE"),
|
||||
)
|
||||
pet_name: Mapped[str] = mapped_column(String(50), nullable=False)
|
||||
|
||||
```
|
||||
|
||||
Note, we are using super powerful SQLAlchemy feature here - Mapped and mapped_column were first introduced in SQLAlchemy 2.0, if this syntax is new for you, read carefully "what's new" part of documentation https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html.
|
||||
|
||||
<br>
|
||||
|
||||
### 2. Create and apply alembic migration
|
||||
|
||||
```bash
|
||||
### Use below commands in root folder in virtualenv ###
|
||||
|
||||
# if you see FAILED: Target database is not up to date.
|
||||
# first use alembic upgrade head
|
||||
|
||||
# Create migration with alembic revision
|
||||
alembic revision --autogenerate -m "create_pet_model"
|
||||
|
||||
|
||||
# File similar to "2022050949_create_pet_model_44b7b689ea5f.py" should appear in `/alembic/versions` folder
|
||||
|
||||
|
||||
# Apply migration using alembic upgrade
|
||||
alembic upgrade head
|
||||
|
||||
# (...)
|
||||
# INFO [alembic.runtime.migration] Running upgrade d1252175c146 -> 44b7b689ea5f, create_pet_model
|
||||
```
|
||||
|
||||
PS. Note, alembic is configured in a way that it work with async setup and also detects specific column changes if using `--autogenerate` flag.
|
||||
|
||||
<br>
|
||||
|
||||
### 3. Create request and response schemas
|
||||
|
||||
There are only 2 files: `requests.py` and `responses.py` in `schemas` folder and I would keep it that way even for few dozen of endpoints. Not to mention this is opinionated.
|
||||
|
||||
```python
|
||||
# app/schemas/requests.py
|
||||
|
||||
(...)
|
||||
|
||||
|
||||
class PetCreateRequest(BaseRequest):
|
||||
pet_name: str
|
||||
|
||||
```
|
||||
|
||||
```python
|
||||
# app/schemas/responses.py
|
||||
|
||||
(...)
|
||||
|
||||
|
||||
class PetResponse(BaseResponse):
|
||||
id: int
|
||||
pet_name: str
|
||||
user_id: str
|
||||
|
||||
```
|
||||
|
||||
<br>
|
||||
|
||||
### 4. Create endpoints
|
||||
|
||||
```python
|
||||
# app/api/endpoints/pets.py
|
||||
|
||||
from fastapi import APIRouter, Depends, status
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import deps
|
||||
from app.models import Pet, User
|
||||
from app.schemas.requests import PetCreateRequest
|
||||
from app.schemas.responses import PetResponse
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.post(
|
||||
"/create",
|
||||
response_model=PetResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
description="Creates new pet. Only for logged users.",
|
||||
)
|
||||
async def create_new_pet(
|
||||
data: PetCreateRequest,
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
current_user: User = Depends(deps.get_current_user),
|
||||
) -> Pet:
|
||||
new_pet = Pet(user_id=current_user.user_id, pet_name=data.pet_name)
|
||||
|
||||
session.add(new_pet)
|
||||
await session.commit()
|
||||
|
||||
return new_pet
|
||||
|
||||
|
||||
@router.get(
|
||||
"/me",
|
||||
response_model=list[PetResponse],
|
||||
status_code=status.HTTP_200_OK,
|
||||
description="Get list of pets for currently logged user.",
|
||||
)
|
||||
async def get_all_my_pets(
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
current_user: User = Depends(deps.get_current_user),
|
||||
) -> list[Pet]:
|
||||
pets = await session.scalars(
|
||||
select(Pet).where(Pet.user_id == current_user.user_id).order_by(Pet.pet_name)
|
||||
)
|
||||
|
||||
return list(pets.all())
|
||||
|
||||
```
|
||||
|
||||
Also, we need to add newly created endpoints to router.
|
||||
|
||||
```python
|
||||
# app/api/api.py
|
||||
|
||||
(...)
|
||||
|
||||
from app.api.endpoints import auth, pets, users
|
||||
|
||||
(...)
|
||||
|
||||
api_router.include_router(pets.router, prefix="/pets", tags=["pets"])
|
||||
|
||||
```
|
||||
|
||||
<br>
|
||||
|
||||
### 5. Write tests
|
||||
|
||||
We will write two really simple tests in combined file inside newly created `app/tests/test_pets` folder.
|
||||
|
||||
```python
|
||||
# app/tests/test_pets/test_pets.py
|
||||
|
||||
from fastapi import status
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.main import app
|
||||
from app.models import Pet, User
|
||||
|
||||
|
||||
async def test_create_new_pet(
|
||||
client: AsyncClient, default_user_headers: dict[str, str], default_user: User
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("create_new_pet"),
|
||||
headers=default_user_headers,
|
||||
json={"pet_name": "Tadeusz"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
result = response.json()
|
||||
assert result["user_id"] == default_user.user_id
|
||||
assert result["pet_name"] == "Tadeusz"
|
||||
|
||||
|
||||
async def test_get_all_my_pets(
|
||||
client: AsyncClient,
|
||||
default_user_headers: dict[str, str],
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
pet1 = Pet(user_id=default_user.user_id, pet_name="Pet_1")
|
||||
pet2 = Pet(user_id=default_user.user_id, pet_name="Pet_2")
|
||||
|
||||
session.add(pet1)
|
||||
session.add(pet2)
|
||||
await session.commit()
|
||||
|
||||
response = await client.get(
|
||||
app.url_path_for("get_all_my_pets"),
|
||||
headers=default_user_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
assert response.json() == [
|
||||
{
|
||||
"user_id": pet1.user_id,
|
||||
"pet_name": pet1.pet_name,
|
||||
"id": pet1.id,
|
||||
},
|
||||
{
|
||||
"user_id": pet2.user_id,
|
||||
"pet_name": pet2.pet_name,
|
||||
"id": pet2.id,
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
```
|
||||
|
||||
## Design
|
||||
|
||||
### Deployment strategies - via Docker image
|
||||
|
||||
This template has by default included `Dockerfile` with [Uvicorn](https://www.uvicorn.org/) webserver, because it's simple and just for showcase purposes, with direct relation to FastAPI and great ease of configuration. You should be able to run container(s) (over :8000 port) and then need to setup the proxy, loadbalancer, with https enbaled, so the app stays behind it.
|
||||
|
||||
If you prefer other webservers for FastAPI, check out [Nginx Unit](https://unit.nginx.org/), [Daphne](https://github.com/django/daphne), [Hypercorn](https://pgjones.gitlab.io/hypercorn/index.html).
|
||||
|
||||
### Docs URL, CORS and Allowed Hosts
|
||||
|
||||
There are some **opinionated** default settings in `/app/main.py` for documentation, CORS and allowed hosts.
|
||||
|
||||
1. Docs
|
||||
|
||||
```python
|
||||
app = FastAPI(
|
||||
title="minimal fastapi postgres template",
|
||||
version="6.0.0",
|
||||
description="https://github.com/rafsaf/minimal-fastapi-postgres-template",
|
||||
openapi_url="/openapi.json",
|
||||
docs_url="/",
|
||||
)
|
||||
```
|
||||
|
||||
Docs page is simpy `/` (by default in FastAPI it is `/docs`). You can change it completely for the project, just as title, version, etc.
|
||||
|
||||
2. CORS
|
||||
|
||||
```python
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=[str(origin) for origin in config.settings.BACKEND_CORS_ORIGINS],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
```
|
||||
|
||||
If you are not sure what are CORS for, follow https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS. React and most frontend frameworks nowadays operate on `http://localhost:3000` thats why it's included in `BACKEND_CORS_ORIGINS` in .env file, before going production be sure to include your frontend domain here, like `https://my-fontend-app.example.com`.
|
||||
|
||||
3. Allowed Hosts
|
||||
|
||||
```python
|
||||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=config.settings.ALLOWED_HOSTS)
|
||||
```
|
||||
|
||||
Prevents HTTP Host Headers attack, you shoud put here you server IP or (preferably) full domain under it's accessible like `example.com`. By default in .env there are two most popular records: `ALLOWED_HOSTS=["localhost", "127.0.0.1"]`
|
||||
|
||||
|
||||
## License
|
||||
|
||||
The code is under MIT License. It's here for educational purposes, created mainly to have a place where up-to-date Python and FastAPI software lives. Do whatever you want with this code.
|
||||
+109
@@ -0,0 +1,109 @@
|
||||
# A generic, single database configuration.
|
||||
|
||||
[alembic]
|
||||
# path to migration scripts
|
||||
script_location = alembic
|
||||
|
||||
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
|
||||
# Uncomment the line below if you want the files to be prepended with date and time
|
||||
file_template = %%(year)d%%(month).2d%%(day).2d%%(minute).2d_%%(slug)s_%%(rev)s
|
||||
|
||||
# sys.path path, will be prepended to sys.path if present.
|
||||
# defaults to the current working directory.
|
||||
prepend_sys_path = .
|
||||
|
||||
# timezone to use when rendering the date within the migration file
|
||||
# as well as the filename.
|
||||
# If specified, requires the python>=3.9 or backports.zoneinfo library.
|
||||
# Any required deps can installed by adding `alembic[tz]` to the pip requirements
|
||||
# string value is passed to ZoneInfo()
|
||||
# leave blank for localtime
|
||||
# timezone =
|
||||
|
||||
# max length of characters to apply to the
|
||||
# "slug" field
|
||||
truncate_slug_length = 40
|
||||
|
||||
# set to 'true' to run the environment during
|
||||
# the 'revision' command, regardless of autogenerate
|
||||
# revision_environment = false
|
||||
|
||||
# set to 'true' to allow .pyc and .pyo files without
|
||||
# a source .py file to be detected as revisions in the
|
||||
# versions/ directory
|
||||
# sourceless = false
|
||||
|
||||
# version location specification; This defaults
|
||||
# to ${script_location}/versions. When using multiple version
|
||||
# directories, initial revisions must be specified with --version-path.
|
||||
# The path separator used here should be the separator specified by "version_path_separator" below.
|
||||
# version_locations = %(here)s/bar:%(here)s/bat:${script_location}/versions
|
||||
|
||||
# version path separator; As mentioned above, this is the character used to split
|
||||
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
|
||||
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
|
||||
# Valid values for version_path_separator are:
|
||||
#
|
||||
# version_path_separator = :
|
||||
# version_path_separator = ;
|
||||
# version_path_separator = space
|
||||
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
|
||||
|
||||
# set to 'true' to search source files recursively
|
||||
# in each "version_locations" directory
|
||||
# new in Alembic version 1.10
|
||||
# recursive_version_locations = false
|
||||
|
||||
# the output encoding used when revision files
|
||||
# are written from script.py.mako
|
||||
# output_encoding = utf-8
|
||||
|
||||
sqlalchemy.url = driver://user:pass@localhost/dbname
|
||||
|
||||
[post_write_hooks]
|
||||
hooks = pre_commit
|
||||
pre_commit.type = console_scripts
|
||||
pre_commit.entrypoint = pre-commit
|
||||
pre_commit.options = run --files REVISION_SCRIPT_FILENAME
|
||||
# This section defines scripts or Python functions that are run
|
||||
# on newly generated revision scripts. See the documentation for further
|
||||
# detail and examples
|
||||
|
||||
# format using "black" - use the console_scripts runner,
|
||||
# against the "black" entrypoint
|
||||
|
||||
|
||||
# Logging configuration
|
||||
[loggers]
|
||||
keys = root,sqlalchemy,alembic
|
||||
|
||||
[handlers]
|
||||
keys = console
|
||||
|
||||
[formatters]
|
||||
keys = generic
|
||||
|
||||
[logger_root]
|
||||
level = WARN
|
||||
handlers = console
|
||||
qualname =
|
||||
|
||||
[logger_sqlalchemy]
|
||||
level = WARN
|
||||
handlers =
|
||||
qualname = sqlalchemy.engine
|
||||
|
||||
[logger_alembic]
|
||||
level = INFO
|
||||
handlers =
|
||||
qualname = alembic
|
||||
|
||||
[handler_console]
|
||||
class = StreamHandler
|
||||
args = (sys.stderr,)
|
||||
level = NOTSET
|
||||
formatter = generic
|
||||
|
||||
[formatter_generic]
|
||||
format = %(levelname)-5.5s [%(name)s] %(message)s
|
||||
datefmt = %H:%M:%S
|
||||
@@ -0,0 +1 @@
|
||||
Generic single-database configuration.
|
||||
@@ -0,0 +1,96 @@
|
||||
import asyncio
|
||||
from logging.config import fileConfig
|
||||
|
||||
from sqlalchemy import Connection, engine_from_config, pool
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
|
||||
from alembic import context
|
||||
from app.core.config import get_settings
|
||||
|
||||
# this is the Alembic Config object, which provides
|
||||
# access to the values within the .ini file in use.
|
||||
config = context.config
|
||||
|
||||
# Interpret the config file for Python logging.
|
||||
# This line sets up loggers basically.
|
||||
fileConfig(config.config_file_name) # type: ignore
|
||||
|
||||
# add your model's MetaData object here
|
||||
# for 'autogenerate' support
|
||||
# from myapp import mymodel
|
||||
# target_metadata = mymodel.Base.metadata
|
||||
from app.models import Base # noqa
|
||||
|
||||
target_metadata = Base.metadata
|
||||
|
||||
# other values from the config, defined by the needs of env.py,
|
||||
# can be acquired:
|
||||
# my_important_option = config.get_main_option("my_important_option")
|
||||
# ... etc.
|
||||
|
||||
|
||||
def get_database_uri() -> str:
|
||||
return get_settings().sqlalchemy_database_uri.render_as_string(hide_password=False)
|
||||
|
||||
|
||||
def run_migrations_offline() -> None:
|
||||
"""Run migrations in 'offline' mode.
|
||||
|
||||
This configures the context with just a URL
|
||||
and not an Engine, though an Engine is acceptable
|
||||
here as well. By skipping the Engine creation
|
||||
we don't even need a DBAPI to be available.
|
||||
|
||||
Calls to context.execute() here emit the given string to the
|
||||
script output.
|
||||
|
||||
"""
|
||||
url = get_database_uri()
|
||||
context.configure(
|
||||
url=url,
|
||||
target_metadata=target_metadata,
|
||||
literal_binds=True,
|
||||
dialect_opts={"paramstyle": "named"},
|
||||
compare_type=True,
|
||||
compare_server_default=True,
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
def do_run_migrations(connection: Connection | None) -> None:
|
||||
context.configure(
|
||||
connection=connection, target_metadata=target_metadata, compare_type=True
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
async def run_migrations_online() -> None:
|
||||
"""Run migrations in 'online' mode.
|
||||
|
||||
In this scenario we need to create an Engine
|
||||
and associate a connection with the context.
|
||||
|
||||
"""
|
||||
configuration = config.get_section(config.config_ini_section)
|
||||
assert configuration
|
||||
configuration["sqlalchemy.url"] = get_database_uri()
|
||||
connectable = AsyncEngine(
|
||||
engine_from_config(
|
||||
configuration,
|
||||
prefix="sqlalchemy.",
|
||||
poolclass=pool.NullPool,
|
||||
future=True,
|
||||
)
|
||||
)
|
||||
async with connectable.connect() as connection:
|
||||
await connection.run_sync(do_run_migrations)
|
||||
|
||||
|
||||
if context.is_offline_mode():
|
||||
run_migrations_offline()
|
||||
else:
|
||||
asyncio.run(run_migrations_online())
|
||||
@@ -0,0 +1,24 @@
|
||||
"""${message}
|
||||
|
||||
Revision ID: ${up_revision}
|
||||
Revises: ${down_revision | comma,n}
|
||||
Create Date: ${create_date}
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
${imports if imports else ""}
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = ${repr(up_revision)}
|
||||
down_revision = ${repr(down_revision)}
|
||||
branch_labels = ${repr(branch_labels)}
|
||||
depends_on = ${repr(depends_on)}
|
||||
|
||||
|
||||
def upgrade():
|
||||
${upgrades if upgrades else "pass"}
|
||||
|
||||
|
||||
def downgrade():
|
||||
${downgrades if downgrades else "pass"}
|
||||
@@ -0,0 +1,83 @@
|
||||
"""init user and refresh token
|
||||
|
||||
Revision ID: c79b0938ea4b
|
||||
Revises:
|
||||
Create Date: 2024-03-03 11:45:21.361225
|
||||
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = "c79b0938ea4b"
|
||||
down_revision = None
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.create_table(
|
||||
"user_account",
|
||||
sa.Column("user_id", sa.Uuid(as_uuid=False), nullable=False),
|
||||
sa.Column("email", sa.String(length=256), nullable=False),
|
||||
sa.Column("hashed_password", sa.String(length=128), nullable=False),
|
||||
sa.Column(
|
||||
"create_time",
|
||||
sa.DateTime(timezone=True),
|
||||
server_default=sa.text("now()"),
|
||||
nullable=False,
|
||||
),
|
||||
sa.Column(
|
||||
"update_time",
|
||||
sa.DateTime(timezone=True),
|
||||
server_default=sa.text("now()"),
|
||||
nullable=False,
|
||||
),
|
||||
sa.PrimaryKeyConstraint("user_id"),
|
||||
)
|
||||
op.create_index(
|
||||
op.f("ix_user_account_email"), "user_account", ["email"], unique=True
|
||||
)
|
||||
op.create_table(
|
||||
"refresh_token",
|
||||
sa.Column("id", sa.BigInteger(), nullable=False),
|
||||
sa.Column("refresh_token", sa.String(length=512), nullable=False),
|
||||
sa.Column("used", sa.Boolean(), nullable=False),
|
||||
sa.Column("exp", sa.BigInteger(), nullable=False),
|
||||
sa.Column("user_id", sa.Uuid(as_uuid=False), nullable=False),
|
||||
sa.Column(
|
||||
"create_time",
|
||||
sa.DateTime(timezone=True),
|
||||
server_default=sa.text("now()"),
|
||||
nullable=False,
|
||||
),
|
||||
sa.Column(
|
||||
"update_time",
|
||||
sa.DateTime(timezone=True),
|
||||
server_default=sa.text("now()"),
|
||||
nullable=False,
|
||||
),
|
||||
sa.ForeignKeyConstraint(
|
||||
["user_id"], ["user_account.user_id"], ondelete="CASCADE"
|
||||
),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
op.create_index(
|
||||
op.f("ix_refresh_token_refresh_token"),
|
||||
"refresh_token",
|
||||
["refresh_token"],
|
||||
unique=True,
|
||||
)
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.drop_index(op.f("ix_refresh_token_refresh_token"), table_name="refresh_token")
|
||||
op.drop_table("refresh_token")
|
||||
op.drop_index(op.f("ix_user_account_email"), table_name="user_account")
|
||||
op.drop_table("user_account")
|
||||
# ### end Alembic commands ###
|
||||
@@ -0,0 +1,6 @@
|
||||
JWT_ERROR_USER_REMOVED = "User removed"
|
||||
PASSWORD_INVALID = "Incorrect email or password"
|
||||
REFRESH_TOKEN_NOT_FOUND = "Refresh token not found"
|
||||
REFRESH_TOKEN_EXPIRED = "Refresh token expired"
|
||||
REFRESH_TOKEN_ALREADY_USED = "Refresh token already used"
|
||||
EMAIL_ADDRESS_ALREADY_USED = "Cannot use this email address"
|
||||
@@ -0,0 +1,34 @@
|
||||
from fastapi import APIRouter
|
||||
|
||||
from app.api import api_messages
|
||||
from app.api.endpoints import auth, users
|
||||
|
||||
auth_router = APIRouter()
|
||||
auth_router.include_router(auth.router, prefix="/auth", tags=["auth"])
|
||||
|
||||
api_router = APIRouter(
|
||||
responses={
|
||||
401: {
|
||||
"description": "No `Authorization` access token header, token is invalid or user removed",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"examples": {
|
||||
"not authenticated": {
|
||||
"summary": "No authorization token header",
|
||||
"value": {"detail": "Not authenticated"},
|
||||
},
|
||||
"invalid token": {
|
||||
"summary": "Token validation failed, decode failed, it may be expired or malformed",
|
||||
"value": {"detail": "Token invalid: {detailed error msg}"},
|
||||
},
|
||||
"removed user": {
|
||||
"summary": api_messages.JWT_ERROR_USER_REMOVED,
|
||||
"value": {"detail": api_messages.JWT_ERROR_USER_REMOVED},
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
api_router.include_router(users.router, prefix="/users", tags=["users"])
|
||||
@@ -0,0 +1,35 @@
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import api_messages
|
||||
from app.core import database_session
|
||||
from app.core.security.jwt import verify_jwt_token
|
||||
from app.models import User
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/access-token")
|
||||
|
||||
|
||||
async def get_session() -> AsyncGenerator[AsyncSession, None]:
|
||||
async with database_session.get_async_session() as session:
|
||||
yield session
|
||||
|
||||
|
||||
async def get_current_user(
|
||||
token: Annotated[str, Depends(oauth2_scheme)],
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> User:
|
||||
token_payload = verify_jwt_token(token)
|
||||
|
||||
user = await session.scalar(select(User).where(User.user_id == token_payload.sub))
|
||||
|
||||
if user is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail=api_messages.JWT_ERROR_USER_REMOVED,
|
||||
)
|
||||
return user
|
||||
@@ -0,0 +1,194 @@
|
||||
import secrets
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordRequestForm
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import api_messages, deps
|
||||
from app.core.config import get_settings
|
||||
from app.core.security.jwt import create_jwt_token
|
||||
from app.core.security.password import (
|
||||
DUMMY_PASSWORD,
|
||||
get_password_hash,
|
||||
verify_password,
|
||||
)
|
||||
from app.models import RefreshToken, User
|
||||
from app.schemas.requests import RefreshTokenRequest, UserCreateRequest
|
||||
from app.schemas.responses import AccessTokenResponse, UserResponse
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
ACCESS_TOKEN_RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
400: {
|
||||
"description": "Invalid email or password",
|
||||
"content": {
|
||||
"application/json": {"example": {"detail": api_messages.PASSWORD_INVALID}}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
REFRESH_TOKEN_RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
400: {
|
||||
"description": "Refresh token expired or is already used",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"examples": {
|
||||
"refresh token expired": {
|
||||
"summary": api_messages.REFRESH_TOKEN_EXPIRED,
|
||||
"value": {"detail": api_messages.REFRESH_TOKEN_EXPIRED},
|
||||
},
|
||||
"refresh token already used": {
|
||||
"summary": api_messages.REFRESH_TOKEN_ALREADY_USED,
|
||||
"value": {"detail": api_messages.REFRESH_TOKEN_ALREADY_USED},
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
404: {
|
||||
"description": "Refresh token does not exist",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"example": {"detail": api_messages.REFRESH_TOKEN_NOT_FOUND}
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@router.post(
|
||||
"/access-token",
|
||||
response_model=AccessTokenResponse,
|
||||
responses=ACCESS_TOKEN_RESPONSES,
|
||||
description="OAuth2 compatible token, get an access token for future requests using username and password",
|
||||
)
|
||||
async def login_access_token(
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
form_data: OAuth2PasswordRequestForm = Depends(),
|
||||
) -> AccessTokenResponse:
|
||||
user = await session.scalar(select(User).where(User.email == form_data.username))
|
||||
|
||||
if user is None:
|
||||
# this is naive method to not return early
|
||||
verify_password(form_data.password, DUMMY_PASSWORD)
|
||||
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=api_messages.PASSWORD_INVALID,
|
||||
)
|
||||
|
||||
if not verify_password(form_data.password, user.hashed_password):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=api_messages.PASSWORD_INVALID,
|
||||
)
|
||||
|
||||
jwt_token = create_jwt_token(user_id=user.user_id)
|
||||
|
||||
refresh_token = RefreshToken(
|
||||
user_id=user.user_id,
|
||||
refresh_token=secrets.token_urlsafe(32),
|
||||
exp=int(time.time() + get_settings().security.refresh_token_expire_secs),
|
||||
)
|
||||
session.add(refresh_token)
|
||||
await session.commit()
|
||||
|
||||
return AccessTokenResponse(
|
||||
access_token=jwt_token.access_token,
|
||||
expires_at=jwt_token.payload.exp,
|
||||
refresh_token=refresh_token.refresh_token,
|
||||
refresh_token_expires_at=refresh_token.exp,
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/refresh-token",
|
||||
response_model=AccessTokenResponse,
|
||||
responses=REFRESH_TOKEN_RESPONSES,
|
||||
description="OAuth2 compatible token, get an access token for future requests using refresh token",
|
||||
)
|
||||
async def refresh_token(
|
||||
data: RefreshTokenRequest,
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
) -> AccessTokenResponse:
|
||||
token = await session.scalar(
|
||||
select(RefreshToken)
|
||||
.where(RefreshToken.refresh_token == data.refresh_token)
|
||||
.with_for_update(skip_locked=True)
|
||||
)
|
||||
|
||||
if token is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=api_messages.REFRESH_TOKEN_NOT_FOUND,
|
||||
)
|
||||
elif time.time() > token.exp:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=api_messages.REFRESH_TOKEN_EXPIRED,
|
||||
)
|
||||
elif token.used:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=api_messages.REFRESH_TOKEN_ALREADY_USED,
|
||||
)
|
||||
|
||||
token.used = True
|
||||
session.add(token)
|
||||
|
||||
jwt_token = create_jwt_token(user_id=token.user_id)
|
||||
|
||||
refresh_token = RefreshToken(
|
||||
user_id=token.user_id,
|
||||
refresh_token=secrets.token_urlsafe(32),
|
||||
exp=int(time.time() + get_settings().security.refresh_token_expire_secs),
|
||||
)
|
||||
session.add(refresh_token)
|
||||
await session.commit()
|
||||
|
||||
return AccessTokenResponse(
|
||||
access_token=jwt_token.access_token,
|
||||
expires_at=jwt_token.payload.exp,
|
||||
refresh_token=refresh_token.refresh_token,
|
||||
refresh_token_expires_at=refresh_token.exp,
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/register",
|
||||
response_model=UserResponse,
|
||||
description="Create new user",
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
)
|
||||
async def register_new_user(
|
||||
new_user: UserCreateRequest,
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
) -> User:
|
||||
user = await session.scalar(select(User).where(User.email == new_user.email))
|
||||
if user is not None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=api_messages.EMAIL_ADDRESS_ALREADY_USED,
|
||||
)
|
||||
|
||||
user = User(
|
||||
email=new_user.email,
|
||||
hashed_password=get_password_hash(new_user.password),
|
||||
)
|
||||
session.add(user)
|
||||
|
||||
try:
|
||||
await session.commit()
|
||||
except IntegrityError: # pragma: no cover
|
||||
await session.rollback()
|
||||
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=api_messages.EMAIL_ADDRESS_ALREADY_USED,
|
||||
)
|
||||
|
||||
return user
|
||||
@@ -0,0 +1,46 @@
|
||||
from fastapi import APIRouter, Depends, status
|
||||
from sqlalchemy import delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import deps
|
||||
from app.core.security.password import get_password_hash
|
||||
from app.models import User
|
||||
from app.schemas.requests import UserUpdatePasswordRequest
|
||||
from app.schemas.responses import UserResponse
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/me", response_model=UserResponse, description="Get current user")
|
||||
async def read_current_user(
|
||||
current_user: User = Depends(deps.get_current_user),
|
||||
) -> User:
|
||||
return current_user
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/me",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
description="Delete current user",
|
||||
)
|
||||
async def delete_current_user(
|
||||
current_user: User = Depends(deps.get_current_user),
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
) -> None:
|
||||
await session.execute(delete(User).where(User.user_id == current_user.user_id))
|
||||
await session.commit()
|
||||
|
||||
|
||||
@router.post(
|
||||
"/reset-password",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
description="Update current user password",
|
||||
)
|
||||
async def reset_current_user_password(
|
||||
user_update_password: UserUpdatePasswordRequest,
|
||||
session: AsyncSession = Depends(deps.get_session),
|
||||
current_user: User = Depends(deps.get_current_user),
|
||||
) -> None:
|
||||
current_user.hashed_password = get_password_hash(user_update_password.password)
|
||||
session.add(current_user)
|
||||
await session.commit()
|
||||
@@ -0,0 +1,70 @@
|
||||
# File with environment variables and general configuration logic.
|
||||
# Env variables are combined in nested groups like "Security", "Database" etc.
|
||||
# So environment variable (case-insensitive) for jwt_secret_key will be "security__jwt_secret_key"
|
||||
#
|
||||
# Pydantic priority ordering:
|
||||
#
|
||||
# 1. (Most important, will overwrite everything) - environment variables
|
||||
# 2. `.env` file in root folder of project
|
||||
# 3. Default values
|
||||
#
|
||||
# "sqlalchemy_database_uri" is computed field that will create valid database URL
|
||||
#
|
||||
# See https://pydantic-docs.helpmanual.io/usage/settings/
|
||||
# Note, complex types like lists are read as json-encoded strings.
|
||||
|
||||
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic import AnyHttpUrl, BaseModel, SecretStr, computed_field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from sqlalchemy.engine.url import URL
|
||||
|
||||
PROJECT_DIR = Path(__file__).parent.parent.parent
|
||||
|
||||
|
||||
class Security(BaseModel):
|
||||
jwt_issuer: str = "my-app"
|
||||
jwt_secret_key: SecretStr
|
||||
jwt_access_token_expire_secs: int = 24 * 3600 # 1d
|
||||
refresh_token_expire_secs: int = 28 * 24 * 3600 # 28d
|
||||
password_bcrypt_rounds: int = 12
|
||||
allowed_hosts: list[str] = ["localhost", "127.0.0.1"]
|
||||
backend_cors_origins: list[AnyHttpUrl] = []
|
||||
|
||||
|
||||
class Database(BaseModel):
|
||||
hostname: str = "postgres"
|
||||
username: str = "postgres"
|
||||
password: SecretStr
|
||||
port: int = 5432
|
||||
db: str = "postgres"
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
security: Security
|
||||
database: Database
|
||||
|
||||
@computed_field # type: ignore[misc]
|
||||
@property
|
||||
def sqlalchemy_database_uri(self) -> URL:
|
||||
return URL.create(
|
||||
drivername="postgresql+asyncpg",
|
||||
username=self.database.username,
|
||||
password=self.database.password.get_secret_value(),
|
||||
host=self.database.hostname,
|
||||
port=self.database.port,
|
||||
database=self.database.db,
|
||||
)
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=f"{PROJECT_DIR}/.env",
|
||||
case_sensitive=False,
|
||||
env_nested_delimiter="__",
|
||||
)
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def get_settings() -> Settings:
|
||||
return Settings() # type: ignore
|
||||
@@ -0,0 +1,36 @@
|
||||
# SQLAlchemy async engine and sessions tools
|
||||
#
|
||||
# https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
|
||||
#
|
||||
# for pool size configuration:
|
||||
# https://docs.sqlalchemy.org/en/20/core/pooling.html#sqlalchemy.pool.Pool
|
||||
|
||||
|
||||
from sqlalchemy.engine.url import URL
|
||||
from sqlalchemy.ext.asyncio import (
|
||||
AsyncEngine,
|
||||
AsyncSession,
|
||||
async_sessionmaker,
|
||||
create_async_engine,
|
||||
)
|
||||
|
||||
from app.core.config import get_settings
|
||||
|
||||
|
||||
def new_async_engine(uri: URL) -> AsyncEngine:
|
||||
return create_async_engine(
|
||||
uri,
|
||||
pool_pre_ping=True,
|
||||
pool_size=5,
|
||||
max_overflow=10,
|
||||
pool_timeout=30.0,
|
||||
pool_recycle=600,
|
||||
)
|
||||
|
||||
|
||||
_ASYNC_ENGINE = new_async_engine(get_settings().sqlalchemy_database_uri)
|
||||
_ASYNC_SESSIONMAKER = async_sessionmaker(_ASYNC_ENGINE, expire_on_commit=False)
|
||||
|
||||
|
||||
def get_async_session() -> AsyncSession: # pragma: no cover
|
||||
return _ASYNC_SESSIONMAKER()
|
||||
@@ -0,0 +1,69 @@
|
||||
import time
|
||||
|
||||
import jwt
|
||||
from fastapi import HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
|
||||
from app.core.config import get_settings
|
||||
|
||||
JWT_ALGORITHM = "HS256"
|
||||
|
||||
|
||||
# Payload follows RFC 7519
|
||||
# https://www.rfc-editor.org/rfc/rfc7519#section-4.1
|
||||
class JWTTokenPayload(BaseModel):
|
||||
iss: str
|
||||
sub: str
|
||||
exp: int
|
||||
iat: int
|
||||
|
||||
|
||||
class JWTToken(BaseModel):
|
||||
payload: JWTTokenPayload
|
||||
access_token: str
|
||||
|
||||
|
||||
def create_jwt_token(user_id: str) -> JWTToken:
|
||||
iat = int(time.time())
|
||||
exp = iat + get_settings().security.jwt_access_token_expire_secs
|
||||
|
||||
token_payload = JWTTokenPayload(
|
||||
iss=get_settings().security.jwt_issuer,
|
||||
sub=user_id,
|
||||
exp=exp,
|
||||
iat=iat,
|
||||
)
|
||||
|
||||
access_token = jwt.encode(
|
||||
token_payload.model_dump(),
|
||||
key=get_settings().security.jwt_secret_key.get_secret_value(),
|
||||
algorithm=JWT_ALGORITHM,
|
||||
)
|
||||
|
||||
return JWTToken(payload=token_payload, access_token=access_token)
|
||||
|
||||
|
||||
def verify_jwt_token(token: str) -> JWTTokenPayload:
|
||||
# Pay attention to verify_signature passed explicite, even if it is the default.
|
||||
# Verification is based on expected payload fields like "exp", "iat" etc.
|
||||
# so if you rename for example "exp" to "my_custom_exp", this is gonna break,
|
||||
# jwt.ExpiredSignatureError will not be raised, that can potentialy
|
||||
# be major security risk - not validating tokens at all.
|
||||
# If unsure, jump into jwt.decode code, make sure tests are passing
|
||||
# https://pyjwt.readthedocs.io/en/stable/usage.html#encoding-decoding-tokens-with-hs256
|
||||
|
||||
try:
|
||||
raw_payload = jwt.decode(
|
||||
token,
|
||||
get_settings().security.jwt_secret_key.get_secret_value(),
|
||||
algorithms=[JWT_ALGORITHM],
|
||||
options={"verify_signature": True},
|
||||
issuer=get_settings().security.jwt_issuer,
|
||||
)
|
||||
except jwt.InvalidTokenError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail=f"Token invalid: {e}",
|
||||
)
|
||||
|
||||
return JWTTokenPayload(**raw_payload)
|
||||
@@ -0,0 +1,19 @@
|
||||
import bcrypt
|
||||
|
||||
from app.core.config import get_settings
|
||||
|
||||
|
||||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||
return bcrypt.checkpw(
|
||||
plain_password.encode("utf-8"), hashed_password.encode("utf-8")
|
||||
)
|
||||
|
||||
|
||||
def get_password_hash(password: str) -> str:
|
||||
return bcrypt.hashpw(
|
||||
password.encode(),
|
||||
bcrypt.gensalt(get_settings().security.password_bcrypt_rounds),
|
||||
).decode()
|
||||
|
||||
|
||||
DUMMY_PASSWORD = get_password_hash("")
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.middleware.trustedhost import TrustedHostMiddleware
|
||||
|
||||
from app.api.api_router import api_router, auth_router
|
||||
from app.core.config import get_settings
|
||||
|
||||
app = FastAPI(
|
||||
title="minimal fastapi postgres template",
|
||||
version="6.0.0",
|
||||
description="https://github.com/rafsaf/minimal-fastapi-postgres-template",
|
||||
openapi_url="/openapi.json",
|
||||
docs_url="/",
|
||||
)
|
||||
|
||||
app.include_router(auth_router)
|
||||
app.include_router(api_router)
|
||||
|
||||
# Sets all CORS enabled origins
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=[
|
||||
str(origin) for origin in get_settings().security.backend_cors_origins
|
||||
],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# Guards against HTTP Host Header attacks
|
||||
app.add_middleware(
|
||||
TrustedHostMiddleware,
|
||||
allowed_hosts=get_settings().security.allowed_hosts,
|
||||
)
|
||||
@@ -0,0 +1,57 @@
|
||||
# SQL Alchemy models declaration.
|
||||
# https://docs.sqlalchemy.org/en/20/orm/quickstart.html#declare-models
|
||||
# mapped_column syntax from SQLAlchemy 2.0.
|
||||
|
||||
# https://alembic.sqlalchemy.org/en/latest/tutorial.html
|
||||
# Note, it is used by alembic migrations logic, see `alembic/env.py`
|
||||
|
||||
# Alembic shortcuts:
|
||||
# # create migration
|
||||
# alembic revision --autogenerate -m "migration_name"
|
||||
|
||||
# # apply all migrations
|
||||
# alembic upgrade head
|
||||
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import BigInteger, Boolean, DateTime, ForeignKey, String, Uuid, func
|
||||
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
|
||||
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
create_time: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now()
|
||||
)
|
||||
update_time: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
|
||||
)
|
||||
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "user_account"
|
||||
|
||||
user_id: Mapped[str] = mapped_column(
|
||||
Uuid(as_uuid=False), primary_key=True, default=lambda _: str(uuid.uuid4())
|
||||
)
|
||||
email: Mapped[str] = mapped_column(
|
||||
String(256), nullable=False, unique=True, index=True
|
||||
)
|
||||
hashed_password: Mapped[str] = mapped_column(String(128), nullable=False)
|
||||
refresh_tokens: Mapped[list["RefreshToken"]] = relationship(back_populates="user")
|
||||
|
||||
|
||||
class RefreshToken(Base):
|
||||
__tablename__ = "refresh_token"
|
||||
|
||||
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
|
||||
refresh_token: Mapped[str] = mapped_column(
|
||||
String(512), nullable=False, unique=True, index=True
|
||||
)
|
||||
used: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
||||
exp: Mapped[int] = mapped_column(BigInteger, nullable=False)
|
||||
user_id: Mapped[str] = mapped_column(
|
||||
ForeignKey("user_account.user_id", ondelete="CASCADE"),
|
||||
)
|
||||
user: Mapped["User"] = relationship(back_populates="refresh_tokens")
|
||||
@@ -0,0 +1,19 @@
|
||||
from pydantic import BaseModel, EmailStr
|
||||
|
||||
|
||||
class BaseRequest(BaseModel):
|
||||
# may define additional fields or config shared across requests
|
||||
pass
|
||||
|
||||
|
||||
class RefreshTokenRequest(BaseRequest):
|
||||
refresh_token: str
|
||||
|
||||
|
||||
class UserUpdatePasswordRequest(BaseRequest):
|
||||
password: str
|
||||
|
||||
|
||||
class UserCreateRequest(BaseRequest):
|
||||
email: EmailStr
|
||||
password: str
|
||||
@@ -0,0 +1,18 @@
|
||||
from pydantic import BaseModel, ConfigDict, EmailStr
|
||||
|
||||
|
||||
class BaseResponse(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
class AccessTokenResponse(BaseResponse):
|
||||
token_type: str = "Bearer"
|
||||
access_token: str
|
||||
expires_at: int
|
||||
refresh_token: str
|
||||
refresh_token_expires_at: int
|
||||
|
||||
|
||||
class UserResponse(BaseResponse):
|
||||
user_id: str
|
||||
email: EmailStr
|
||||
@@ -0,0 +1,135 @@
|
||||
import asyncio
|
||||
import os
|
||||
from collections.abc import AsyncGenerator, Generator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
import sqlalchemy
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import (
|
||||
AsyncSession,
|
||||
async_sessionmaker,
|
||||
)
|
||||
|
||||
from app.core import database_session
|
||||
from app.core.config import get_settings
|
||||
from app.core.security.jwt import create_jwt_token
|
||||
from app.core.security.password import get_password_hash
|
||||
from app.main import app as fastapi_app
|
||||
from app.models import Base, User
|
||||
|
||||
default_user_id = "b75365d9-7bf9-4f54-add5-aeab333a087b"
|
||||
default_user_email = "geralt@wiedzmin.pl"
|
||||
default_user_password = "geralt"
|
||||
default_user_access_token = create_jwt_token(default_user_id).access_token
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", autouse=True)
|
||||
async def fixture_setup_new_test_database() -> None:
|
||||
worker_name = os.getenv("PYTEST_XDIST_WORKER", "gw0")
|
||||
test_db_name = f"test_db_{worker_name}"
|
||||
|
||||
# create new test db using connection to current database
|
||||
conn = await database_session._ASYNC_ENGINE.connect()
|
||||
await conn.execution_options(isolation_level="AUTOCOMMIT")
|
||||
await conn.execute(sqlalchemy.text(f"DROP DATABASE IF EXISTS {test_db_name}"))
|
||||
await conn.execute(sqlalchemy.text(f"CREATE DATABASE {test_db_name}"))
|
||||
await conn.close()
|
||||
|
||||
session_mpatch = pytest.MonkeyPatch()
|
||||
session_mpatch.setenv("DATABASE__DB", test_db_name)
|
||||
session_mpatch.setenv("SECURITY__PASSWORD_BCRYPT_ROUNDS", "4")
|
||||
|
||||
# force settings to use now monkeypatched environments
|
||||
get_settings.cache_clear()
|
||||
|
||||
# monkeypatch test database engine
|
||||
engine = database_session.new_async_engine(get_settings().sqlalchemy_database_uri)
|
||||
|
||||
session_mpatch.setattr(
|
||||
database_session,
|
||||
"_ASYNC_ENGINE",
|
||||
engine,
|
||||
)
|
||||
session_mpatch.setattr(
|
||||
database_session,
|
||||
"_ASYNC_SESSIONMAKER",
|
||||
async_sessionmaker(engine, expire_on_commit=False),
|
||||
)
|
||||
|
||||
# create app tables in test database
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function", autouse=True)
|
||||
async def fixture_clean_get_settings_between_tests() -> AsyncGenerator[None, None]:
|
||||
yield
|
||||
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(name="default_hashed_password", scope="session")
|
||||
async def fixture_default_hashed_password() -> str:
|
||||
return get_password_hash(default_user_password)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(name="session", scope="function")
|
||||
async def fixture_session_with_rollback(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> AsyncGenerator[AsyncSession, None]:
|
||||
# we want to monkeypatch get_async_session with one bound to session
|
||||
# that we will always rollback on function scope
|
||||
|
||||
connection = await database_session._ASYNC_ENGINE.connect()
|
||||
transaction = await connection.begin()
|
||||
|
||||
session = AsyncSession(bind=connection, expire_on_commit=False)
|
||||
|
||||
monkeypatch.setattr(
|
||||
database_session,
|
||||
"get_async_session",
|
||||
lambda: session,
|
||||
)
|
||||
|
||||
yield session
|
||||
|
||||
await session.close()
|
||||
await transaction.rollback()
|
||||
await connection.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(name="client", scope="function")
|
||||
async def fixture_client(session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
|
||||
transport = ASGITransport(app=fastapi_app) # type: ignore
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as aclient:
|
||||
aclient.headers.update({"Host": "localhost"})
|
||||
yield aclient
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(name="default_user", scope="function")
|
||||
async def fixture_default_user(
|
||||
session: AsyncSession, default_hashed_password: str
|
||||
) -> User:
|
||||
default_user = User(
|
||||
user_id=default_user_id,
|
||||
email=default_user_email,
|
||||
hashed_password=default_hashed_password,
|
||||
)
|
||||
session.add(default_user)
|
||||
await session.commit()
|
||||
await session.refresh(default_user)
|
||||
return default_user
|
||||
|
||||
|
||||
@pytest.fixture(name="default_user_headers", scope="function")
|
||||
def fixture_default_user_headers(default_user: User) -> dict[str, str]:
|
||||
return {"Authorization": f"Bearer {default_user_access_token}"}
|
||||
@@ -0,0 +1,66 @@
|
||||
import pytest
|
||||
from fastapi import routing, status
|
||||
from freezegun import freeze_time
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import api_messages
|
||||
from app.api.api_router import api_router
|
||||
from app.core.security.jwt import create_jwt_token
|
||||
from app.models import User
|
||||
|
||||
|
||||
@pytest.mark.parametrize("api_route", api_router.routes)
|
||||
async def test_api_routes_raise_401_on_jwt_decode_errors(
|
||||
client: AsyncClient,
|
||||
api_route: routing.APIRoute,
|
||||
) -> None:
|
||||
for method in api_route.methods:
|
||||
response = await client.request(
|
||||
method=method,
|
||||
url=api_route.path,
|
||||
headers={"Authorization": "Bearer garbage-invalid-jwt"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
assert response.json() == {"detail": "Token invalid: Not enough segments"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("api_route", api_router.routes)
|
||||
async def test_api_routes_raise_401_on_jwt_expired_token(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
api_route: routing.APIRoute,
|
||||
) -> None:
|
||||
with freeze_time("2023-01-01"):
|
||||
jwt = create_jwt_token(default_user.user_id)
|
||||
with freeze_time("2023-02-01"):
|
||||
for method in api_route.methods:
|
||||
response = await client.request(
|
||||
method=method,
|
||||
url=api_route.path,
|
||||
headers={"Authorization": f"Bearer {jwt.access_token}"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
assert response.json() == {"detail": "Token invalid: Signature has expired"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("api_route", api_router.routes)
|
||||
async def test_api_routes_raise_401_on_jwt_user_deleted(
|
||||
client: AsyncClient,
|
||||
default_user_headers: dict[str, str],
|
||||
default_user: User,
|
||||
api_route: routing.APIRoute,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
await session.execute(delete(User).where(User.user_id == default_user.user_id))
|
||||
await session.commit()
|
||||
|
||||
for method in api_route.methods:
|
||||
response = await client.request(
|
||||
method=method,
|
||||
url=api_route.path,
|
||||
headers=default_user_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
assert response.json() == {"detail": api_messages.JWT_ERROR_USER_REMOVED}
|
||||
@@ -0,0 +1,193 @@
|
||||
import time
|
||||
|
||||
from fastapi import status
|
||||
from freezegun import freeze_time
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import api_messages
|
||||
from app.core.config import get_settings
|
||||
from app.core.security.jwt import verify_jwt_token
|
||||
from app.main import app
|
||||
from app.models import RefreshToken, User
|
||||
from app.tests.conftest import default_user_password
|
||||
|
||||
|
||||
async def test_login_access_token_has_response_status_code(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
|
||||
async def test_login_access_token_jwt_has_valid_token_type(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
assert token["token_type"] == "Bearer"
|
||||
|
||||
|
||||
@freeze_time("2023-01-01")
|
||||
async def test_login_access_token_jwt_has_valid_expire_time(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
current_timestamp = int(time.time())
|
||||
assert (
|
||||
token["expires_at"]
|
||||
== current_timestamp + get_settings().security.jwt_access_token_expire_secs
|
||||
)
|
||||
|
||||
|
||||
@freeze_time("2023-01-01")
|
||||
async def test_login_access_token_returns_valid_jwt_access_token(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
now = int(time.time())
|
||||
token = response.json()
|
||||
token_payload = verify_jwt_token(token["access_token"])
|
||||
|
||||
assert token_payload.sub == default_user.user_id
|
||||
assert token_payload.iat == now
|
||||
assert token_payload.exp == token["expires_at"]
|
||||
|
||||
|
||||
async def test_login_access_token_refresh_token_has_valid_expire_time(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
current_time = int(time.time())
|
||||
assert (
|
||||
token["refresh_token_expires_at"]
|
||||
== current_time + get_settings().security.refresh_token_expire_secs
|
||||
)
|
||||
|
||||
|
||||
async def test_login_access_token_refresh_token_exists_in_db(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
|
||||
token_db_count = await session.scalar(
|
||||
select(func.count()).where(RefreshToken.refresh_token == token["refresh_token"])
|
||||
)
|
||||
assert token_db_count == 1
|
||||
|
||||
|
||||
async def test_login_access_token_refresh_token_in_db_has_valid_fields(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": default_user_password,
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
result = await session.scalars(
|
||||
select(RefreshToken).where(RefreshToken.refresh_token == token["refresh_token"])
|
||||
)
|
||||
refresh_token = result.one()
|
||||
|
||||
assert refresh_token.user_id == default_user.user_id
|
||||
assert refresh_token.exp == token["refresh_token_expires_at"]
|
||||
assert not refresh_token.used
|
||||
|
||||
|
||||
async def test_auth_access_token_fail_for_not_existing_user_with_message(
|
||||
client: AsyncClient,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": "non-existing",
|
||||
"password": "bla",
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json() == {"detail": api_messages.PASSWORD_INVALID}
|
||||
|
||||
|
||||
async def test_auth_access_token_fail_for_invalid_password_with_message(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("login_access_token"),
|
||||
data={
|
||||
"username": default_user.email,
|
||||
"password": "invalid",
|
||||
},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json() == {"detail": api_messages.PASSWORD_INVALID}
|
||||
@@ -0,0 +1,272 @@
|
||||
import time
|
||||
|
||||
from fastapi import status
|
||||
from freezegun import freeze_time
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import api_messages
|
||||
from app.core.config import get_settings
|
||||
from app.core.security.jwt import verify_jwt_token
|
||||
from app.main import app
|
||||
from app.models import RefreshToken, User
|
||||
|
||||
|
||||
async def test_refresh_token_fails_with_message_when_token_does_not_exist(
|
||||
client: AsyncClient,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json() == {"detail": api_messages.REFRESH_TOKEN_NOT_FOUND}
|
||||
|
||||
|
||||
async def test_refresh_token_fails_with_message_when_token_is_expired(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) - 1,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json() == {"detail": api_messages.REFRESH_TOKEN_EXPIRED}
|
||||
|
||||
|
||||
async def test_refresh_token_fails_with_message_when_token_is_used(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=True,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json() == {"detail": api_messages.REFRESH_TOKEN_ALREADY_USED}
|
||||
|
||||
|
||||
async def test_refresh_token_success_response_status_code(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
|
||||
async def test_refresh_token_success_old_token_is_used(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
used_test_refresh_token = await session.scalar(
|
||||
select(RefreshToken).where(RefreshToken.refresh_token == "blaxx")
|
||||
)
|
||||
assert used_test_refresh_token is not None
|
||||
assert used_test_refresh_token.used
|
||||
|
||||
|
||||
async def test_refresh_token_success_jwt_has_valid_token_type(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
assert token["token_type"] == "Bearer"
|
||||
|
||||
|
||||
@freeze_time("2023-01-01")
|
||||
async def test_refresh_token_success_jwt_has_valid_expire_time(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
current_timestamp = int(time.time())
|
||||
assert (
|
||||
token["expires_at"]
|
||||
== current_timestamp + get_settings().security.jwt_access_token_expire_secs
|
||||
)
|
||||
|
||||
|
||||
@freeze_time("2023-01-01")
|
||||
async def test_refresh_token_success_jwt_has_valid_access_token(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
now = int(time.time())
|
||||
token = response.json()
|
||||
token_payload = verify_jwt_token(token["access_token"])
|
||||
|
||||
assert token_payload.sub == default_user.user_id
|
||||
assert token_payload.iat == now
|
||||
assert token_payload.exp == token["expires_at"]
|
||||
|
||||
|
||||
@freeze_time("2023-01-01")
|
||||
async def test_refresh_token_success_refresh_token_has_valid_expire_time(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
current_time = int(time.time())
|
||||
assert (
|
||||
token["refresh_token_expires_at"]
|
||||
== current_time + get_settings().security.refresh_token_expire_secs
|
||||
)
|
||||
|
||||
|
||||
async def test_refresh_token_success_new_refresh_token_is_in_db(
|
||||
client: AsyncClient,
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
test_refresh_token = RefreshToken(
|
||||
user_id=default_user.user_id,
|
||||
refresh_token="blaxx",
|
||||
exp=int(time.time()) + 1000,
|
||||
used=False,
|
||||
)
|
||||
session.add(test_refresh_token)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("refresh_token"),
|
||||
json={
|
||||
"refresh_token": "blaxx",
|
||||
},
|
||||
)
|
||||
|
||||
token = response.json()
|
||||
token_db_count = await session.scalar(
|
||||
select(func.count()).where(RefreshToken.refresh_token == token["refresh_token"])
|
||||
)
|
||||
assert token_db_count == 1
|
||||
@@ -0,0 +1,63 @@
|
||||
from fastapi import status
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api import api_messages
|
||||
from app.main import app
|
||||
from app.models import User
|
||||
|
||||
|
||||
async def test_register_new_user_status_code(
|
||||
client: AsyncClient,
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("register_new_user"),
|
||||
json={
|
||||
"email": "test@email.com",
|
||||
"password": "testtesttest",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
|
||||
async def test_register_new_user_creates_record_in_db(
|
||||
client: AsyncClient,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
await client.post(
|
||||
app.url_path_for("register_new_user"),
|
||||
json={
|
||||
"email": "test@email.com",
|
||||
"password": "testtesttest",
|
||||
},
|
||||
)
|
||||
|
||||
user_count = await session.scalar(
|
||||
select(func.count()).where(User.email == "test@email.com")
|
||||
)
|
||||
assert user_count == 1
|
||||
|
||||
|
||||
async def test_register_new_user_cannot_create_already_created_user(
|
||||
client: AsyncClient,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
user = User(
|
||||
email="test@email.com",
|
||||
hashed_password="bla",
|
||||
)
|
||||
session.add(user)
|
||||
await session.commit()
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("register_new_user"),
|
||||
json={
|
||||
"email": "test@email.com",
|
||||
"password": "testtesttest",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json() == {"detail": api_messages.EMAIL_ADDRESS_ALREADY_USED}
|
||||
@@ -0,0 +1,84 @@
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from freezegun import freeze_time
|
||||
from pydantic import SecretStr
|
||||
|
||||
from app.core.config import get_settings
|
||||
from app.core.security import jwt
|
||||
|
||||
|
||||
def test_jwt_access_token_can_be_decoded_back_into_user_id() -> None:
|
||||
user_id = "test_user_id"
|
||||
token = jwt.create_jwt_token(user_id)
|
||||
|
||||
payload = jwt.verify_jwt_token(token=token.access_token)
|
||||
assert payload.sub == user_id
|
||||
|
||||
|
||||
@freeze_time("2024-01-01")
|
||||
def test_jwt_payload_is_correct() -> None:
|
||||
user_id = "test_user_id"
|
||||
token = jwt.create_jwt_token(user_id)
|
||||
|
||||
assert token.payload.iat == int(time.time())
|
||||
assert token.payload.sub == user_id
|
||||
assert token.payload.iss == get_settings().security.jwt_issuer
|
||||
assert (
|
||||
token.payload.exp
|
||||
== int(time.time()) + get_settings().security.jwt_access_token_expire_secs
|
||||
)
|
||||
|
||||
|
||||
def test_jwt_error_after_exp_time() -> None:
|
||||
user_id = "test_user_id"
|
||||
with freeze_time("2024-01-01"):
|
||||
token = jwt.create_jwt_token(user_id)
|
||||
with freeze_time("2024-02-01"):
|
||||
with pytest.raises(HTTPException) as e:
|
||||
jwt.verify_jwt_token(token=token.access_token)
|
||||
|
||||
assert e.value.detail == "Token invalid: Signature has expired"
|
||||
|
||||
|
||||
def test_jwt_error_before_iat_time() -> None:
|
||||
user_id = "test_user_id"
|
||||
with freeze_time("2024-01-01"):
|
||||
token = jwt.create_jwt_token(user_id)
|
||||
with freeze_time("2023-12-01"):
|
||||
with pytest.raises(HTTPException) as e:
|
||||
jwt.verify_jwt_token(token=token.access_token)
|
||||
|
||||
assert e.value.detail == "Token invalid: The token is not yet valid (iat)"
|
||||
|
||||
|
||||
def test_jwt_error_with_invalid_token() -> None:
|
||||
with pytest.raises(HTTPException) as e:
|
||||
jwt.verify_jwt_token(token="invalid!")
|
||||
|
||||
assert e.value.detail == "Token invalid: Not enough segments"
|
||||
|
||||
|
||||
def test_jwt_error_with_invalid_issuer() -> None:
|
||||
user_id = "test_user_id"
|
||||
token = jwt.create_jwt_token(user_id)
|
||||
|
||||
get_settings().security.jwt_issuer = "another_issuer"
|
||||
|
||||
with pytest.raises(HTTPException) as e:
|
||||
jwt.verify_jwt_token(token=token.access_token)
|
||||
|
||||
assert e.value.detail == "Token invalid: Invalid issuer"
|
||||
|
||||
|
||||
def test_jwt_error_with_invalid_secret_key() -> None:
|
||||
user_id = "test_user_id"
|
||||
token = jwt.create_jwt_token(user_id)
|
||||
|
||||
get_settings().security.jwt_secret_key = SecretStr("the secret has changed now!")
|
||||
|
||||
with pytest.raises(HTTPException) as e:
|
||||
jwt.verify_jwt_token(token=token.access_token)
|
||||
|
||||
assert e.value.detail == "Token invalid: Signature verification failed"
|
||||
@@ -0,0 +1,11 @@
|
||||
from app.core.security.password import get_password_hash, verify_password
|
||||
|
||||
|
||||
def test_hashed_password_is_verified() -> None:
|
||||
pwd_hash = get_password_hash("my_password")
|
||||
assert verify_password("my_password", pwd_hash)
|
||||
|
||||
|
||||
def test_invalid_password_is_not_verified() -> None:
|
||||
pwd_hash = get_password_hash("my_password")
|
||||
assert not verify_password("my_password_invalid", pwd_hash)
|
||||
@@ -0,0 +1,36 @@
|
||||
from fastapi import status
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.main import app
|
||||
from app.models import User
|
||||
|
||||
|
||||
async def test_delete_current_user_status_code(
|
||||
client: AsyncClient,
|
||||
default_user_headers: dict[str, str],
|
||||
) -> None:
|
||||
response = await client.delete(
|
||||
app.url_path_for("delete_current_user"),
|
||||
headers=default_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
|
||||
async def test_delete_current_user_is_deleted_in_db(
|
||||
client: AsyncClient,
|
||||
default_user_headers: dict[str, str],
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
await client.delete(
|
||||
app.url_path_for("delete_current_user"),
|
||||
headers=default_user_headers,
|
||||
)
|
||||
|
||||
user = await session.scalar(
|
||||
select(User).where(User.user_id == default_user.user_id)
|
||||
)
|
||||
assert user is None
|
||||
@@ -0,0 +1,33 @@
|
||||
from fastapi import status
|
||||
from httpx import AsyncClient
|
||||
|
||||
from app.main import app
|
||||
from app.tests.conftest import (
|
||||
default_user_email,
|
||||
default_user_id,
|
||||
)
|
||||
|
||||
|
||||
async def test_read_current_user_status_code(
|
||||
client: AsyncClient, default_user_headers: dict[str, str]
|
||||
) -> None:
|
||||
response = await client.get(
|
||||
app.url_path_for("read_current_user"),
|
||||
headers=default_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
|
||||
async def test_read_current_user_response(
|
||||
client: AsyncClient, default_user_headers: dict[str, str]
|
||||
) -> None:
|
||||
response = await client.get(
|
||||
app.url_path_for("read_current_user"),
|
||||
headers=default_user_headers,
|
||||
)
|
||||
|
||||
assert response.json() == {
|
||||
"user_id": default_user_id,
|
||||
"email": default_user_email,
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
from fastapi import status
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.security.password import verify_password
|
||||
from app.main import app
|
||||
from app.models import User
|
||||
|
||||
|
||||
async def test_reset_current_user_password_status_code(
|
||||
client: AsyncClient,
|
||||
default_user_headers: dict[str, str],
|
||||
) -> None:
|
||||
response = await client.post(
|
||||
app.url_path_for("reset_current_user_password"),
|
||||
headers=default_user_headers,
|
||||
json={"password": "test_pwd"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
|
||||
async def test_reset_current_user_password_is_changed_in_db(
|
||||
client: AsyncClient,
|
||||
default_user_headers: dict[str, str],
|
||||
default_user: User,
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
await client.post(
|
||||
app.url_path_for("reset_current_user_password"),
|
||||
headers=default_user_headers,
|
||||
json={"password": "test_pwd"},
|
||||
)
|
||||
|
||||
user = await session.scalar(
|
||||
select(User).where(User.user_id == default_user.user_id)
|
||||
)
|
||||
assert user is not None
|
||||
assert verify_password("test_pwd", user.hashed_password)
|
||||
@@ -0,0 +1,23 @@
|
||||
# For local development, only database is running
|
||||
#
|
||||
# docker compose up -d
|
||||
# uvicorn app.main:app --reload
|
||||
#
|
||||
|
||||
services:
|
||||
postgres_db:
|
||||
restart: unless-stopped
|
||||
image: postgres:16
|
||||
volumes:
|
||||
- postgres_db:/var/lib/postgresql/data
|
||||
environment:
|
||||
- POSTGRES_DB=${DATABASE__DB}
|
||||
- POSTGRES_USER=${DATABASE__USERNAME}
|
||||
- POSTGRES_PASSWORD=${DATABASE__PASSWORD}
|
||||
env_file:
|
||||
- .env
|
||||
ports:
|
||||
- "${DATABASE__PORT}:5432"
|
||||
|
||||
volumes:
|
||||
postgres_db:
|
||||
@@ -0,0 +1,4 @@
|
||||
#!/bin/bash
|
||||
|
||||
echo "Run migrations"
|
||||
alembic upgrade head
|
||||
Generated
+1780
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,60 @@
|
||||
[tool.poetry]
|
||||
authors = ["admin <admin@example.com>"]
|
||||
description = "FastAPI project generated using minimal-fastapi-postgres-template."
|
||||
name = "app"
|
||||
version = "0.1.0-alpha"
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python = "^3.12"
|
||||
|
||||
alembic = "^1.13.1"
|
||||
asyncpg = "^0.29.0"
|
||||
bcrypt = "^4.1.2"
|
||||
fastapi = "^0.110.1"
|
||||
pydantic = {extras = ["dotenv", "email"], version = "^2.6.4"}
|
||||
pydantic-settings = "^2.2.1"
|
||||
pyjwt = "^2.8.0"
|
||||
python-multipart = "^0.0.9"
|
||||
sqlalchemy = "^2.0.29"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
coverage = "^7.4.4"
|
||||
freezegun = "^1.4.0"
|
||||
gevent = "^24.2.1"
|
||||
httpx = "^0.27.0"
|
||||
mypy = "^1.9.0"
|
||||
pre-commit = "^3.7.0"
|
||||
pytest = "^8.1.1"
|
||||
# do not bump pytest-asyncio until https://github.com/pytest-dev/pytest-asyncio/issues/706 resolved
|
||||
pytest-asyncio = "0.21.1"
|
||||
pytest-cov = "^5.0.0"
|
||||
pytest-xdist = "^3.5.0"
|
||||
ruff = "^0.3.5"
|
||||
types-passlib = "^1.7.7.20240327"
|
||||
uvicorn = {extras = ["standard"], version = "^0.29.0"}
|
||||
|
||||
[build-system]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
requires = ["poetry-core>=1.0.0"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
addopts = "-vv -n auto --cov --cov-report xml --cov-report term-missing --cov-fail-under=100"
|
||||
asyncio_mode = "auto"
|
||||
testpaths = ["app/tests"]
|
||||
|
||||
[tool.coverage.run]
|
||||
concurrency = ["gevent"]
|
||||
omit = ["app/tests/*"]
|
||||
source = ["app"]
|
||||
|
||||
[tool.mypy]
|
||||
python_version = "3.12"
|
||||
strict = true
|
||||
|
||||
[tool.ruff]
|
||||
target-version = "py312"
|
||||
|
||||
[tool.ruff.lint]
|
||||
# pycodestyle, pyflakes, isort, pylint, pyupgrade
|
||||
ignore = ["E501"]
|
||||
select = ["E", "F", "I", "PL", "UP", "W"]
|
||||
Reference in New Issue
Block a user