Commit addd3900 authored by jonas's avatar jonas

Add task validation

parent e2616f66
.PHONY: clean install install-dev flake8 pylint lint test
.PHONY: clean flake8 pylint lint test
# Run all commands in a single shell
# See https://www.gnu.org/software/make/manual/html_node/One-Shell.html
......@@ -27,35 +27,30 @@ clean:
venv:
test -f venv/bin/activate || python3 -m venv venv
install: venv
source venv/bin/activate
${CMD_PIP_INSTALL} pip
${CMD_PIP_INSTALL} wheel
${CMD_PIP_INSTALL} -r requirements.txt
install-dev: install
source venv/bin/activate
${CMD_PIP_INSTALL} flake8 pylint pytest pytest-cov pytest-xdist
flake8: install-dev
flake8: venv
source venv/bin/activate
${CMD_FLAKE8} ${TEST_MODULE}
${CMD_FLAKE8} ${APP_MODULE}
pylint: install-dev
pylint: venv
source venv/bin/activate
${CMD_PYLINT} ${TEST_MODULE}
${CMD_PYLINT} ${APP_MODULE}
lint: flake8 pylint
test: install-dev
test: venv
source venv/bin/activate
${CMD_PYTEST}
test-failed: install-dev
test-failed: venv
source venv/bin/activate
${CMD_PYTEST} --lf
all: clean install-dev lint test
all: clean venv lint test
......@@ -24,20 +24,18 @@ from app.api.users_tokens import (
users_tokens_delete,
users_tokens_get_all,
)
from app.api.users_tasks import (
users_tasks_add_multiple,
users_tasks_add,
users_tasks_delete,
users_tasks_get_all,
users_tasks_get,
users_tasks_update,
from app.api.users_works import (
users_works_delete,
users_works_get_all,
users_works_get_elapsed,
users_works_get,
users_works_set,
)
from app.api.users_timetables import (
users_timetables_add,
users_timetables_delete,
users_timetables_get_all,
users_timetables_get_current,
users_timetables_get_all_elapsed,
users_timetables_get,
users_timetables_update,
)
......@@ -45,50 +43,55 @@ from app.api.system import system_visualization_setup, system_visualization_expo
api = Blueprint("api", __name__)
DELETE = "DELETE"
GET = "GET"
PATCH = "PATCH"
POST = "POST"
PUT = "PUT"
routes = [
# Auth
R("/auth/login", "POST", login),
R("/auth/logout", "POST", logout),
R("/auth/login", POST, login),
R("/auth/logout", POST, logout),
# Users
R("/users", "GET", users_get_all),
R("/users", "POST", users_add),
R("/users/<user_id>", "DELETE", users_delete),
R("/users/<user_id>", "GET", users_get),
R("/users/<user_id>", "PATCH", users_update),
R("/users", GET, users_get_all),
R("/users", POST, users_add),
R("/users/<user_id>", DELETE, users_delete),
R("/users/<user_id>", GET, users_get),
R("/users/<user_id>", PATCH, users_update),
# User Tokens
R("/users/<user_id>/tokens", "GET", users_tokens_get_all),
R("/users/<user_id>/tokens", "POST", users_tokens_add),
R("/users/<user_id>/tokens/<token_id>", "DELETE", users_tokens_delete),
# User Tasks
R("/users/<user_id>/tasks", "GET", users_tasks_get_all),
R("/users/<user_id>/tasks", "POST", users_tasks_add),
R("/users/<user_id>/tasks/<task_id>", "DELETE", users_tasks_delete),
R("/users/<user_id>/tasks/<task_id>", "GET", users_tasks_get),
R("/users/<user_id>/tasks/<task_id>", "PATCH", users_tasks_update),
R("/users/<user_id>/tasks/multiple", "POST", users_tasks_add_multiple),
R("/users/<user_id>/tokens", GET, users_tokens_get_all),
R("/users/<user_id>/tokens", POST, users_tokens_add),
R("/users/<user_id>/tokens/<token_id>", DELETE, users_tokens_delete),
# User work
R("/users/<user_id>/works", GET, users_works_get_all),
R("/users/<user_id>/works/elapsed", GET, users_works_get_elapsed),
R("/users/<user_id>/works/<activity_id>", DELETE, users_works_delete),
R("/users/<user_id>/works/<activity_id>", GET, users_works_get),
R("/users/<user_id>/works/<activity_id>", PUT, users_works_set),
# User timetables
R("/users/<user_id>/timetables", "GET", users_timetables_get_all),
R("/users/<user_id>/timetables", "POST", users_timetables_add),
R("/users/<user_id>/timetables/<timetable_id>", "DELETE", users_timetables_delete),
R("/users/<user_id>/timetables/<timetable_id>", "GET", users_timetables_get),
R("/users/<user_id>/timetables/<timetable_id>", "PATCH", users_timetables_update),
R("/users/<user_id>/timetables/current", "GET", users_timetables_get_current),
R("/users/<user_id>/timetables/elapsed", "GET", users_timetables_get_all_elapsed),
R("/users/<user_id>/timetables", GET, users_timetables_get_all),
R("/users/<user_id>/timetables", POST, users_timetables_add),
R("/users/<user_id>/timetables/current", GET, users_timetables_get_current),
R("/users/<user_id>/timetables/<timetable_id>", DELETE, users_timetables_delete),
R("/users/<user_id>/timetables/<timetable_id>", GET, users_timetables_get),
R("/users/<user_id>/timetables/<timetable_id>", PATCH, users_timetables_update),
# Tags
R("/tags", "GET", tags_get_all),
R("/tags", "POST", tags_add),
R("/tags/<tag_id>", "DELETE", tags_delete),
R("/tags/<tag_id>", "GET", tags_get),
R("/tags/<tag_id>", "PATCH", tags_update),
R("/tags", GET, tags_get_all),
R("/tags", POST, tags_add),
R("/tags/<tag_id>", DELETE, tags_delete),
R("/tags/<tag_id>", GET, tags_get),
R("/tags/<tag_id>", PATCH, tags_update),
# Activities
R("/activities", "GET", activities_get_all),
R("/activities", "POST", activities_add),
R("/activities/<activity_id>", "DELETE", activities_delete),
R("/activities/<activity_id>", "GET", activities_get),
R("/activities/<activity_id>", "PATCH", activities_update),
R("/activities", GET, activities_get_all),
R("/activities", POST, activities_add),
R("/activities/<activity_id>", DELETE, activities_delete),
R("/activities/<activity_id>", GET, activities_get),
R("/activities/<activity_id>", PATCH, activities_update),
# System
R("/system/visualization/setup", "GET", system_visualization_setup),
R("/system/visualization/export", "GET", system_visualization_export),
R("/system/visualization/setup", GET, system_visualization_setup),
R("/system/visualization/export", GET, system_visualization_export),
]
for route in routes:
......
......@@ -5,7 +5,7 @@ from voluptuous import Schema
from app.api.helpers import check_none
from app.api.schemas import activity_schema
from app.decorators import dataschema
from app.decorators import jsonschema
from app.helpers import APIResult
from app.jwt import jwt_role_required
from app.models.activities import Activity
......@@ -22,13 +22,13 @@ def activities_get(activity_id):
@jwt_role_required("admin")
@dataschema(Schema(activity_schema, required=True))
@jsonschema(Schema(activity_schema, required=True))
def activities_add(name, description, tags):
return APIResult(Activity(name, description, tags).add(), status=201)
@jwt_role_required("admin")
@dataschema(Schema(activity_schema))
@jsonschema(Schema(activity_schema))
def activities_update(activity_id, **kwargs):
def update_func(obj):
for key, value in kwargs.items():
......
......@@ -5,13 +5,13 @@ from voluptuous import Schema
from app.api.schemas import auth_schema
from app.bcrypt import check_password_hash
from app.decorators import dataschema
from app.decorators import jsonschema
from app.helpers import APIResult, APIException
from app.models.tokens import Token
from app.models.users import User
@dataschema(Schema(auth_schema, required=True))
@jsonschema(Schema(auth_schema, required=True))
def login(username, password):
user = User.get_by_username(username)
......
# -*- coding: utf-8 -*-
from datetime import date
from voluptuous import All, Length, Email, Match, In, Clamp, Optional
from voluptuous import All, Length, Email, Match, In, Range, Optional
from app.models.helpers import (
LENGTH_NAME_MAX,
......@@ -11,6 +11,7 @@ from app.models.helpers import (
LENGTH_TEXT_MAX,
LENGTH_TEXT_MIN,
REGEX_USERNAME,
ROLES,
)
fragment_username = All(
......@@ -18,17 +19,14 @@ fragment_username = All(
Match(REGEX_USERNAME, "Invalid username"),
Length(min=LENGTH_NAME_MIN, max=LENGTH_NAME_MAX),
)
fragment_password = All(str, Length(min=LENGTH_PASSWORD_MIN, max=LENGTH_PASSWORD_MAX))
fragment_roles = [All(str, In(ROLES))]
fragment_day_date = All(date.fromisoformat)
fragment_work_time = All(int, Range(min=0, max=(24 * 60 * 60)))
fragment_work_day = {"day_date": fragment_day_date, "work_time": fragment_work_time}
# TODO: Improve roles validation and list # pylint: disable=fixme
fragment_roles = [All(str, In(["admin"]))]
fragment_day_time = All(int, Clamp(min=0, max=(24 * 60 * 60)))
fragment_day_date = All(
lambda v: date.fromisoformat(v) # pylint: disable=unnecessary-lambda
)
# day_date_schema = {"day_date": fragment_day_date}
# work_time_schema = {"work_time": fragment_work_time}
tag_schema = {"name": All(str, Length(min=LENGTH_NAME_MIN, max=LENGTH_NAME_MAX))}
......@@ -53,26 +51,16 @@ activity_schema = {
"tags": All([tag_schema], Length(max=5)),
}
task_schema = {
"activity_id": int,
"day_date": fragment_day_date,
"working_time": fragment_day_time,
}
task_multiple_schema = {
"start_date": fragment_day_date,
"end_date": fragment_day_date,
"activities": [{"activity_id": int, "working_time": fragment_day_time}],
}
work_schema = {"work_days": [fragment_work_day]}
timetable_schema = {
"start_date": fragment_day_date,
"end_date": fragment_day_date,
Optional("monday"): fragment_day_time,
Optional("tuesday"): fragment_day_time,
Optional("wednesday"): fragment_day_time,
Optional("thursday"): fragment_day_time,
Optional("friday"): fragment_day_time,
Optional("saturday"): fragment_day_time,
Optional("sunday"): fragment_day_time,
Optional("monday"): fragment_work_time,
Optional("tuesday"): fragment_work_time,
Optional("wednesday"): fragment_work_time,
Optional("thursday"): fragment_work_time,
Optional("friday"): fragment_work_time,
Optional("saturday"): fragment_work_time,
Optional("sunday"): fragment_work_time,
}
......@@ -2,7 +2,7 @@
from app.helpers import APIResult, APIException
from app.jwt import jwt_role_required
from app.models.tasks import Task
from app.models.works import Work
from app.visualization import visualization
......@@ -15,7 +15,7 @@ def system_visualization_setup():
@jwt_role_required("admin")
def system_visualization_export():
try:
visualization.export(Task.get_all())
visualization.export(Work.get_all())
except Exception as e:
raise APIException(str(e), status=500)
return APIResult(status=200)
......@@ -5,7 +5,7 @@ from voluptuous import Schema
from app.api.helpers import check_none
from app.api.schemas import tag_schema
from app.decorators import dataschema
from app.decorators import jsonschema
from app.helpers import APIResult
from app.jwt import jwt_role_required
from app.models.tags import Tag
......@@ -22,13 +22,13 @@ def tags_get(tag_id):
@jwt_role_required("admin")
@dataschema(Schema(tag_schema, required=True))
@jsonschema(Schema(tag_schema, required=True))
def tags_add(name):
return APIResult(Tag(name).add(), status=201)
@jwt_role_required("admin")
@dataschema(Schema(tag_schema))
@jsonschema(Schema(tag_schema))
def tags_update(tag_id, **kwargs):
def update_func(obj):
for key, value in kwargs.items():
......
......@@ -5,7 +5,7 @@ from flask_jwt_extended import get_jwt_identity
from app.api.helpers import check_none
from app.api.schemas import user_schema
from app.decorators import dataschema
from app.decorators import jsonschema
from app.helpers import APIResult, APIException
from app.jwt import jwt_role_required
from app.models.tokens import Token
......@@ -23,13 +23,13 @@ def users_get(user_id):
@jwt_role_required("admin")
@dataschema(Schema(user_schema, required=True))
@jsonschema(Schema(user_schema, required=True))
def users_add(username, password, email, tags, roles):
return APIResult(User(username, password, email, tags, roles).add(), status=201)
@jwt_role_required("admin", ownership_check=True)
@dataschema(Schema(user_schema))
@jsonschema(Schema(user_schema))
def users_update(user_id, **kwargs):
identity = get_jwt_identity()
......
# -*- coding: utf-8 -*-
from voluptuous import Schema
from app.api.helpers import check_none
from app.api.schemas import task_schema, task_multiple_schema
from app.decorators import dataschema
from app.helpers import APIResult
from app.jwt import jwt_role_required
from app.models.tasks import Task
@jwt_role_required("admin", ownership_check=True)
def users_tasks_get_all(user_id):
return APIResult(Task.get_all_by_user(user_id))
@jwt_role_required("admin", ownership_check=True)
def users_tasks_get(user_id, task_id):
return APIResult(check_none(Task, Task.get_by_user(user_id, task_id)))
@jwt_role_required("admin", ownership_check=True)
@dataschema(Schema(task_schema, required=True))
def users_tasks_add(user_id, activity_id, day_date, working_time):
return APIResult(
Task(activity_id, day_date, working_time, user_id).add(), status=201
)
@jwt_role_required("admin", ownership_check=True)
@dataschema(Schema(task_multiple_schema, required=True))
def users_tasks_add_multiple(user_id, start_date, end_date, activities):
return APIResult(
Task.add_multiple(user_id, start_date, end_date, activities), status=201
)
@jwt_role_required("admin", ownership_check=True)
@dataschema(Schema(task_schema))
def users_tasks_update(user_id, task_id, **kwargs):
def update_func(obj):
for key, value in kwargs.items():
setattr(obj, key, value)
return APIResult(
check_none(Task, Task.get_by_user(user_id, task_id)).update(update_func)
)
@jwt_role_required("admin", ownership_check=True)
def users_tasks_delete(user_id, task_id):
check_none(Task, check_none(Task, Task.get_by_user(user_id, task_id))).delete()
return APIResult()
......@@ -4,11 +4,10 @@ from voluptuous import Schema
from app.api.helpers import check_none
from app.api.schemas import timetable_schema
from app.decorators import dataschema
from app.decorators import jsonschema
from app.helpers import APIResult
from app.jwt import jwt_role_required
from app.models.timetables import Timetable
from app.models.elapsed_working_time import ElapsedWorkingTime
@jwt_role_required("admin", ownership_check=True)
......@@ -28,21 +27,14 @@ def users_timetables_get_current(user_id):
return APIResult(check_none(Timetable, Timetable.get_current_by_user(user_id)))
@jwt_role_required("admin", ownership_check=True)
def users_timetables_get_all_elapsed(user_id):
return APIResult(
check_none(ElapsedWorkingTime, ElapsedWorkingTime.get_all_by_user(user_id))
)
@jwt_role_required("admin")
@dataschema(Schema(timetable_schema))
@jsonschema(Schema(timetable_schema))
def users_timetables_add(user_id, start_date, end_date, **days):
return APIResult(Timetable(user_id, start_date, end_date, **days).add(), status=201)
@jwt_role_required("admin")
@dataschema(Schema(timetable_schema))
@jsonschema(Schema(timetable_schema))
def users_timetables_update(user_id, timetable_id, **kwargs):
def update_func(obj):
for key, value in kwargs.items():
......
......@@ -4,7 +4,7 @@ from voluptuous import Schema
from app.api.helpers import check_none
from app.api.schemas import token_schema
from app.decorators import dataschema
from app.decorators import jsonschema
from app.helpers import APIResult
from app.jwt import jwt_role_required
from app.models.tokens import Token
......@@ -17,7 +17,7 @@ def users_tokens_get_all(user_id):
@jwt_role_required("admin", ownership_check=True)
@dataschema(Schema(token_schema, required=True))
@jsonschema(Schema(token_schema, required=True))
def users_tokens_add(user_id, label, expires):
user = check_none(User, User.get(user_id))
......
# -*- coding: utf-8 -*-
from voluptuous import Schema
from app.api.helpers import check_none
from app.api.schemas import work_schema
from app.decorators import jsonschema
from app.helpers import APIResult
from app.jwt import jwt_role_required
from app.models.works import Work, ElapsedWorkDay
@jwt_role_required("admin", ownership_check=True)
def users_works_get_all(user_id):
return APIResult(Work.get_all_by_user(user_id))
@jwt_role_required("admin", ownership_check=True)
def users_works_get(user_id, activity_id):
return APIResult(check_none(Work, Work.get_by_user(user_id, activity_id)))
@jwt_role_required("admin", ownership_check=True)
def users_works_get_elapsed(user_id):
return APIResult(ElapsedWorkDay.get_all_by_user(user_id))
@jwt_role_required("admin", ownership_check=True)
@jsonschema(Schema(work_schema, required=True))
def users_works_set(user_id, activity_id, work_days):
return APIResult(Work(user_id, activity_id, work_days).set(), status=200)
@jwt_role_required("admin", ownership_check=True)
def users_works_delete(user_id, activity_id):
check_none(Work, check_none(Work, Work.get_by_user(user_id, activity_id))).delete()
return APIResult()
......@@ -7,10 +7,10 @@ from voluptuous import Invalid, MultipleInvalid
from app.helpers import APIException
def dataschema(schema):
def jsonschema(schema):
"""
dataschema decorator will validate the given schema and pass the parsed
data into the attached function arguments.
jsonschema decorator will validate the given schema and pass the parsed
json body into the attached function arguments.
If the parsed data is a list, please iterate over the args argument, if it
is a single object, each values will be passed as arguments using kwargs.
......
......@@ -20,7 +20,7 @@ class Activity(BaseMixin, db.Model):
name = db.Column(db.String(LENGTH_NAME_MAX), unique=True, nullable=False)
description = db.Column(db.String(LENGTH_TEXT_MAX))
__tags = db.relationship(
"Tag", secondary=activities_tags, backref=db.backref("activities")
"Tag", secondary=activities_tags, backref="activities", order_by="Tag.name"
)
@property
......
# -*- coding: utf-8 -*-
from app.models.helpers import db, BaseMixin, sanitize_date
class ElapsedWorkingTime(BaseMixin, db.Model):
__tablename__ = "elapsed_working_time"
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), primary_key=True)
timetable_id = db.Column(db.Integer, db.ForeignKey("timetables.id"))
day_date = db.Column(db.Date, primary_key=True)
working_time = db.Column(db.Integer, nullable=False)
def __init__(self, user_id, timetable_id, day_date, working_time):
self.user_id = user_id
self.timetable_id = timetable_id
self.day_date = sanitize_date(day_date)
self.working_time = working_time
def to_dict(self):
return {
"created": self.created,
"updated": self.updated,
"user_id": self.user_id,
"timetable_id": self.timetable_id,
"day_date": self.day_date,
"working_time": self.working_time,
}
@classmethod
def get_all_by_user(cls, user_id):
return cls.query.filter_by(user_id=user_id).order_by(cls.day_date.asc()).all()
@classmethod
def get_all_between_date_by_user(cls, user_id, start_date, end_date):
return (
cls.query.filter_by(user_id=user_id)
.filter(start_date <= cls.day_date, cls.day_date <= end_date)
.order_by(cls.day_date.asc())
.all()
)
@classmethod
def get_by_user(cls, user_id, day_date):
return (
cls.query.filter_by(user_id=user_id)
.filter_by(day_date=day_date)
.one_or_none()
)
# -*- coding: utf-8 -*-
from datetime import datetime, date, timedelta
from datetime import datetime, date
from flask import json
from app.db import db
......@@ -17,16 +17,13 @@ LENGTH_EMAIL_MAX = 255
REGEX_USERNAME = "^[a-zA-Z0-9_.-]+$"
ROLES = ["admin"]
def sanitize_date(d):
return date.fromisoformat(d) if isinstance(d, str) else d
def daterange(start_date, end_date):
for n in range(int((end_date - start_date).days)):
yield start_date + timedelta(n)
def isoweekday_string(d):
days = {
1: "monday",
......@@ -71,6 +68,9 @@ class BaseMixin:
db.session.commit()
return self
def delete(self):
def delete(self, commit=True):
db.session.delete(self)
db.session.commit()
if commit:
db.session.commit()
else:
db.session.flush()
......@@ -34,7 +34,3 @@ class Tag(BaseMixin, db.Model):
return Tag(new_tag["name"])
return list(map(to_obj, tags))
@classmethod
def get_by_name(cls, name):
return cls.query.filter_by(name=name).one_or_none()
# -*- coding: utf-8 -*-
from sqlalchemy import inspect
from sqlalchemy.orm import lazyload
from app.helpers import APIException
from app.models.helpers import db, BaseMixin, sanitize_date, daterange
from app.models.timetables import Timetable
from app.models.elapsed_working_time import ElapsedWorkingTime
class Task(BaseMixin, db.Model):
__tablename__ = "tasks"
id = db.Column(db.Integer, primary_key=True)
activity_id = db.Column(db.Integer, db.ForeignKey("activities.id"), nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
day_date = db.Column(db.Date, nullable=False)
working_time = db.Column(db.Integer, nullable=False)
activity = db.relationship("Activity")
user = db.relationship("User")
def __init__(self, activity_id, day_date, working_time=0, user_id=None):
self.activity_id = activity_id
self.day_date = sanitize_date(day_date)
self.working_time = working_time
if user_id is not None:
self.user_id = user_id
def to_dict(self):
return {
"id": self.id,
"created": self.created,
"updated": self.updated,
"user_id": self.user_id,
"activity": self.activity,
"day_date": self.day_date,
"working_time": self.working_time,
}
@classmethod
def get_all(cls):
return cls.query.options(lazyload(cls.user), lazyload(cls.activity)).all()
@classmethod
def get_all_by_user(cls, user_id):
return cls.query.filter_by(user_id=user_id).all()
@classmethod
def get_by_user(cls, user_id, task_id):
return cls.query.filter_by(user_id=user_id).filter_by(id=task_id).one_or_none()
def add(self):
# Get user timetable for the task date and extract working time for the task day
timetable = Timetable.get_at_date_by_user(self.user_id, self.day_date)