ulinja/fastapi-svelte-template
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(backend): make list endpoints sortable

Browse source
This commit is contained in:
Julian Lobbes 2023-05-22 13:44:10 +02:00
parent bd191cd420
commit e4061b1654
9 changed files with 101 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
backend/todo/crud/todos.py
View file

@ -8,6 +8,7 @@ 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
def create_todo(db: Session, todo: todoschema.TodoItemCreate, user_id: int) -> todoschema.TodoItem:
@ -39,7 +40,13 @@ def read_todo(db: Session, todo_id: int) -> todoschema.TodoItem:
return todoschema.TodoItem.from_orm(db_todo)
def read_todos_for_user(db: Session, user_id: int, skip: int = 0, limit: int = 100) -> list[todoschema.TodoItem]:
def read_todos_for_user(
db: Session,
user_id: int,
skip: int = 0, limit: int = 100,
sortby: todomodel.SortableTodoItemField = todomodel.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."""
for parameter in [skip, limit]:
@ -52,7 +59,7 @@ def read_todos_for_user(db: Session, user_id: int, skip: int = 0, limit: int = 1
if not db_user:
raise NotFoundException(f"User with id '{user_id}' not found.")
db_todos = db.query(todomodel.TodoItem).filter(todomodel.TodoItem.user_id == user_id).offset(skip).limit(limit).all()
db_todos = db.query(todomodel.TodoItem).filter(todomodel.TodoItem.user_id == user_id).order_by(sortorder.call(sortby.field)).offset(skip).limit(limit).all()
return [todoschema.TodoItem.from_orm(db_todo) for db_todo in db_todos]

10
backend/todo/crud/users.py
View file

@ -6,6 +6,7 @@ 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.utils.exceptions import NotFoundException, InvalidFilterParameterException
@ -56,7 +57,12 @@ def read_user_by_email(db: Session, email: str) -> userschema.User:
return userschema.User.from_orm(db_user)
def read_users(db: Session, skip: int = 0, limit: int = 100) -> list[userschema.User]:
def read_users(
db: Session,
skip: int = 0, limit: int = 100,
sortby: usermodel.SortableUserField = usermodel.SortableUserField('id'),
sortorder: SortOrder = SortOrder['asc'],
) -> list[userschema.User]:
"""Returns an range of users from the database."""
for parameter in [skip, limit]:
@ -65,7 +71,7 @@ def read_users(db: Session, skip: int = 0, limit: int = 100) -> list[userschema.
if parameter < 0:
raise InvalidFilterParameterException(f"Parameter '{parameter}' cannot be smaller than zero.")
db_users = db.query(usermodel.User).offset(skip).limit(limit).all()
db_users = db.query(usermodel.User).order_by(sortorder.call(sortby.field)).offset(skip).limit(limit).all()
return [userschema.User.from_orm(db_user) for db_user in db_users]

4
backend/todo/database/migrations/versions/335e07a98bc8_create_user_tables.py
View file

@ -21,12 +21,12 @@ def upgrade() -> None:
op.create_table(
'users',
Column('id', Integer, primary_key=True, autoincrement=True, index=True),
Column('email', String, unique=True, nullable=False),
Column('email', String, unique=True, nullable=False, index=True),
Column('password', String, nullable=False),
Column('created', DateTime(timezone=True), nullable=False, server_default=now()),
Column('updated', DateTime(timezone=True), nullable=False, server_default=now(), onupdate=now()),
Column('first_name', String, nullable=False),
Column('last_name', String, nullable=False),
Column('last_name', String, nullable=False, index=True),
)

6
backend/todo/database/migrations/versions/7e5a8cabd3a4_create_todo_tables.py
View file

@ -22,13 +22,13 @@ def upgrade() -> None:
op.create_table(
'todo_items',
Column('id', Integer, primary_key=True, autoincrement=True, index=True),
Column('title', String, nullable=False),
Column('title', String, nullable=False, index=True),
Column('description', String, nullable=False),
Column('done', Boolean, nullable=False, default=False, index=True),
Column('created', DateTime(timezone=True), nullable=False, server_default=now()),
Column('updated', DateTime(timezone=True), nullable=False, server_default=now(), onupdate=now()),
Column('updated', DateTime(timezone=True), nullable=False, server_default=now(), onupdate=now(), index=True),
Column('finished', DateTime(timezone=True), nullable=True, default=None),
Column('user_id', Integer, ForeignKey('users.id', ondelete="CASCADE"), nullable=False),
Column('user_id', Integer, ForeignKey('users.id', ondelete="CASCADE"), nullable=False, index=True),
)

26
backend/todo/models/common.py Normal file
View file

@ -0,0 +1,26 @@
"""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.")

21
backend/todo/models/todos.py
View file

@ -16,13 +16,28 @@ class TodoItem(Base):
__tablename__ = "todo_items"
id = Column('id', Integer, primary_key=True, autoincrement=True, index=True)
title = Column('title', String, nullable=False)
title = Column('title', String, nullable=False, index=True)
description = Column('description', String, nullable=False)
done = Column('done', Boolean, nullable=False, default=False, index=True)
created = Column('created', DateTime(timezone=True), nullable=False, server_default=now())
updated = Column('updated', DateTime(timezone=True), nullable=False, server_default=now(), onupdate=now())
updated = Column('updated', DateTime(timezone=True), nullable=False, server_default=now(), onupdate=now(), index=True)
finished = Column('finished', DateTime(timezone=True), nullable=True, default=None)
user_id = Column('user_id', Integer, ForeignKey('users.id', ondelete="CASCADE"), nullable=False)
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)

19
backend/todo/models/users.py
View file

@ -15,11 +15,26 @@ class User(Base):
__tablename__ = "users"
id = Column('id', Integer, primary_key=True, autoincrement=True, index=True)
email = Column('email', String, unique=True, nullable=False)
email = Column('email', String, unique=True, nullable=False, index=True)
password = Column('password', String, nullable=False)
created = Column('created', DateTime(timezone=True), nullable=False, server_default=now())
updated = Column('updated', DateTime(timezone=True), nullable=False, server_default=now(), onupdate=now())
first_name = Column('first_name', String, nullable=False)
last_name = Column('last_name', String, nullable=False)
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)

10
backend/todo/routes/todos.py
View file

@ -8,6 +8,8 @@ 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
router = APIRouter(
@ -38,7 +40,13 @@ 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, db: Session = Depends(get_db)):
def read_todos(
user_id: int,
skip: int = 0, limit: int = 100,
sortby: SortableTodoItemField = SortableTodoItemField['updated'],
sortorder: SortOrder = SortOrder['desc'],
db: Session = Depends(get_db)
):
try:
return todocrud.read_todos_for_user(db=db, user_id=user_id, skip=skip, limit=limit)
except InvalidFilterParameterException as e:

11
backend/todo/routes/users.py
View file

@ -8,6 +8,8 @@ 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
router = APIRouter(
@ -40,9 +42,14 @@ 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, db: Session = Depends(get_db)):
def read_users(
skip: int = 0, limit: int = 100,
sortby: SortableUserField = SortableUserField['id'],
sortorder: SortOrder = SortOrder['asc'],
db: Session = Depends(get_db)
):
try:
return usercrud.read_users(db=db, skip=skip, limit=limit)
return usercrud.read_users(db=db, skip=skip, limit=limit, sortby=sortby, sortorder=sortorder)
except InvalidFilterParameterException as e:
raise HTTPException(400, fmt(str(e)))