refactor: fastapi dependency injections and consistent route naming

This commit is contained in:
Julian Lobbes 2023-05-30 15:27:29 +01:00
parent 80a2ab0319
commit 3c4a7bacdf
18 changed files with 172 additions and 80 deletions

View file

@ -8,7 +8,8 @@ from todo.models import todos as todomodel
from todo.models import users as usermodel
from todo.schemas import todos as todoschema
from todo.utils.exceptions import NotFoundException, InvalidFilterParameterException
from todo.models.common import SortOrder
from todo.dependencies.common import SortOrder
from todo.dependencies.todos import SortableTodoItemField
def create_todo(db: Session, todo: todoschema.TodoItemCreate, user_id: int) -> todoschema.TodoItem:
@ -44,7 +45,7 @@ def read_todos_for_user(
db: Session,
user_id: int,
skip: int = 0, limit: int = 100,
sortby: todomodel.SortableTodoItemField = todomodel.SortableTodoItemField('updated'),
sortby: SortableTodoItemField = SortableTodoItemField('updated'),
sortorder: SortOrder = SortOrder['desc'],
) -> list[todoschema.TodoItem]:
"""Returns a range of todo-items of the user with the specified user_id from the database."""

View file

@ -6,7 +6,8 @@ from sqlalchemy.orm import Session
from todo.models import users as usermodel
from todo.schemas import users as userschema
from todo.models.common import SortOrder
from todo.dependencies.common import SortOrder
from todo.dependencies.users import SortableUserField
from todo.utils.exceptions import NotFoundException, InvalidFilterParameterException
@ -60,7 +61,7 @@ def read_user_by_email(db: Session, email: str) -> userschema.User:
def read_users(
db: Session,
skip: int = 0, limit: int = 100,
sortby: usermodel.SortableUserField = usermodel.SortableUserField('id'),
sortby: SortableUserField = SortableUserField('id'),
sortorder: SortOrder = SortOrder['asc'],
) -> list[userschema.User]:
"""Returns an range of users from the database."""

View file

@ -0,0 +1,43 @@
"""This module provides FastAPI dependencies for commonly used query parameters."""
from enum import Enum
from typing import Callable
import sqlalchemy
from todo.database.engine import Base
class SortOrder(Enum):
"""Possible sort orders for database queries."""
asc = 'asc'
desc = 'desc'
@property
def call(self) -> Callable:
"""Returns the sqlalchemy sort function depending on the instance value."""
if self.value == 'asc':
return sqlalchemy.asc
elif self.value == 'desc':
return sqlalchemy.desc
else:
raise RuntimeError("Logic error.")
class PaginationParams():
"""Represents query parameters used for pagination, when querying for a list of items.
Attributes
----------
skip : int
The number of items to be skipped from the start of the start of the list.
limit : int
Limits the total number of returned list items to the specified number.
"""
def __init__(self, skip: int = 0, limit: int = 100):
self.skip = skip
self.limit = limit

View file

@ -0,0 +1,35 @@
"""Query parameters used to sort todo-items."""
from enum import Enum
from sqlalchemy import Column
from todo.models.todos import TodoItem
from todo.dependencies.common import PaginationParams, SortOrder
class SortableTodoItemField(Enum):
"""Defines which fields todo-item lists can be sorted on."""
id = 'id'
title = 'title'
done = 'done'
created = 'created'
updated = 'updated'
finished = 'finished'
@property
def field(self) -> Column:
return getattr(TodoItem, self.value)
class TodoItemSortablePaginationParams(PaginationParams):
def __init__(
self,
skip: int = 0, limit: int = 100,
sortby: SortableTodoItemField = SortableTodoItemField['updated'],
sortorder: SortOrder = SortOrder['desc'],
):
super().__init__(skip=skip, limit=limit)
self.sortby = sortby
self.sortorder = sortorder

View file

@ -0,0 +1,35 @@
"""Query parameters used to sort users."""
from enum import Enum
from sqlalchemy import Column
from todo.models.users import User
from todo.dependencies.common import PaginationParams, SortOrder
class SortableUserField(Enum):
"""Defines which fields user lists can be sorted on."""
id = 'id'
email = 'email'
created = 'created'
updated = 'updated'
first_name = 'first_name'
last_name = 'last_name'
@property
def field(self) -> Column:
return getattr(User, self.value)
class UserSortablePaginationParams(PaginationParams):
def __init__(
self,
skip: int = 0, limit: int = 100,
sortby: SortableUserField = SortableUserField['id'],
sortorder: SortOrder = SortOrder['asc'],
):
super().__init__(skip=skip, limit=limit)
self.sortby = sortby
self.sortorder = sortorder

View file

@ -1,26 +0,0 @@
"""This module contains common utilities for handling models."""
import enum
from typing import Callable
import sqlalchemy
class SortOrder(enum.Enum):
"""Possible sort orders for database queries."""
asc = 'asc'
ASC = 'asc'
desc = 'desc'
DESC = 'desc'
@property
def call(self) -> Callable:
"""Returns the sqlalchemy sort function depending on the instance value."""
if self.value == 'asc':
return sqlalchemy.asc
elif self.value == 'desc':
return sqlalchemy.desc
else:
raise RuntimeError("Logic error.")

View file

@ -1,8 +1,6 @@
"""This module defines the SQL data model for todo items."""
import enum
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime, Date, Enum, CheckConstraint, Boolean
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime, Boolean
from sqlalchemy.sql.functions import now
from sqlalchemy.orm import relationship
@ -26,18 +24,3 @@ class TodoItem(Base):
user_id = Column('user_id', Integer, ForeignKey('users.id', ondelete="CASCADE"), nullable=False, index=True)
user = relationship(User, back_populates="todo_items", uselist=False)
class SortableTodoItemField(enum.Enum):
"""Defines which fields todo-item lists can be sorted on."""
id = 'id'
title = 'title'
done = 'done'
created = 'created'
updated = 'updated'
finished = 'finished'
@property
def field(self) -> Column:
return getattr(TodoItem, self.value)

View file

@ -1,7 +1,5 @@
"""This module defines the SQL data model for users."""
import enum
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql.functions import now
from sqlalchemy.orm import relationship
@ -23,18 +21,3 @@ class User(Base):
last_name = Column('last_name', String, nullable=False, index=True)
todo_items = relationship("TodoItem", back_populates="user", uselist=True, cascade="all, delete")
class SortableUserField(enum.Enum):
"""Defines which fields user lists can be sorted on."""
id = 'id'
email = 'email'
created = 'created'
updated = 'updated'
first_name = 'first_name'
last_name = 'last_name'
@property
def field(self) -> Column:
return getattr(User, self.value)

View file

@ -1,5 +1,7 @@
"""This module contains endpoints for operations related to users."""
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
@ -8,12 +10,11 @@ from todo.schemas import todos as todoschema
from todo.crud import todos as todocrud
from todo.utils.exceptions import NotFoundException, InvalidFilterParameterException
from todo.utils.exceptions import create_exception_dict as fmt
from todo.models.todos import SortableTodoItemField
from todo.models.common import SortOrder
from todo.dependencies.todos import TodoItemSortablePaginationParams
router = APIRouter(
prefix="/todo",
prefix="/todos",
tags=["todo-items"]
)
@ -42,13 +43,16 @@ def read_todo(todo_id: int, db: Session = Depends(get_db)):
@router.get("/user/{user_id}", response_model=list[todoschema.TodoItem])
def read_todos(
user_id: int,
skip: int = 0, limit: int = 100,
sortby: SortableTodoItemField = SortableTodoItemField['updated'],
sortorder: SortOrder = SortOrder['desc'],
commons: Annotated[TodoItemSortablePaginationParams, Depends(TodoItemSortablePaginationParams)],
db: Session = Depends(get_db)
):
try:
return todocrud.read_todos_for_user(db=db, user_id=user_id, skip=skip, limit=limit, sortby=sortby, sortorder=sortorder)
return todocrud.read_todos_for_user(
db=db,
user_id=user_id,
skip=commons.skip, limit=commons.limit,
sortby=commons.sortby, sortorder=commons.sortorder,
)
except InvalidFilterParameterException as e:
raise HTTPException(400, fmt(str(e)))
except NotFoundException as e:

View file

@ -1,5 +1,7 @@
"""This module contains endpoints for operations related to users."""
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
@ -8,8 +10,7 @@ from todo.schemas import users as userschema
from todo.crud import users as usercrud
from todo.utils.exceptions import NotFoundException, InvalidFilterParameterException
from todo.utils.exceptions import create_exception_dict as fmt
from todo.models.common import SortOrder
from todo.models.users import SortableUserField
from todo.dependencies.users import UserSortablePaginationParams
router = APIRouter(
@ -43,13 +44,17 @@ def read_user(id: int, db: Session = Depends(get_db)):
@router.get("/", response_model=list[userschema.User])
def read_users(
skip: int = 0, limit: int = 100,
sortby: SortableUserField = SortableUserField['id'],
sortorder: SortOrder = SortOrder['asc'],
commons: Annotated[UserSortablePaginationParams, Depends(UserSortablePaginationParams)],
db: Session = Depends(get_db)
):
try:
return usercrud.read_users(db=db, skip=skip, limit=limit, sortby=sortby, sortorder=sortorder)
return usercrud.read_users(
db=db,
skip=commons.skip,
limit=commons.limit,
sortby=commons.sortby,
sortorder=commons.sortorder
)
except InvalidFilterParameterException as e:
raise HTTPException(400, fmt(str(e)))

View file

@ -0,0 +1,12 @@
"""This module contains common functions for schema validation."""
def is_valid_id(id: int) -> bool:
"""Checks whether the specified id is a valid ID.
Performs a shallow check on whether the ID is a valid primary key.
"""
if not isinstance(id, int):
raise TypeError(f"Expected an integer but got: {type(id)}")
return id > 0

View file

@ -6,6 +6,8 @@ from typing import Optional
from pydantic import BaseModel, validator
from todo.schemas.common import is_valid_id
class AbstractTodoItemValidator(BaseModel, ABC):
"""Base class for todo-item validators shared with child classes."""
@ -55,5 +57,11 @@ class TodoItem(AbstractTodoItem):
updated: datetime
finished: datetime | None
@validator('id')
def assert_id_is_valid(cls, id):
if not is_valid_id(id):
raise ValueError("ID is invalid.")
return id
class Config:
orm_mode = True

View file

@ -7,6 +7,8 @@ from re import compile
from pydantic import BaseModel, validator
from todo.schemas.common import is_valid_id
def is_valid_email_format(input_str: str) -> bool:
"""Checks whether the input string is a valid email address format.
@ -106,5 +108,11 @@ class User(AbstractUser):
created: datetime
updated: datetime
@validator('id')
def assert_id_is_valid(cls, id):
if not is_valid_id(id):
raise ValueError("ID is invalid.")
return id
class Config:
orm_mode = True

View file

@ -44,8 +44,8 @@
const table = {
caption: `List of TODOs for user ${data.user.first_name} ${data.user.last_name}`,
columns: cols,
itemsEndpoint: `/api/todo/user/${data.user.id}/`,
itemCountEndpoint: `/api/todo/user/${data.user.id}/total/`,
itemsEndpoint: `/api/todos/user/${data.user.id}/`,
itemCountEndpoint: `/api/todos/user/${data.user.id}/total/`,
};
</script>