taskflower/src/taskflower/sanitize/task.py

166 lines
No EOL
3.7 KiB
Python

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Self
import humanize
from taskflower.auth.permission import NPT, NamespacePermissionType
from taskflower.auth.permission.checks import assert_user_perms_on_task
from taskflower.auth.permission.lookups import get_user_perms_on_task
from taskflower.db import db, db_fetch_by_id
from taskflower.db.model.tag import Tag, TagToTask, TaskCategory
from taskflower.db.model.task import Task
from taskflower.db.model.user import User
from taskflower.types.either import Either, gather_successes
from taskflower.util.time import ensure_timezone_aware, now
def _due_str(due: datetime|None) -> str:
if not due:
return 'N/A'
due = ensure_timezone_aware(due)
cur_dt = now()
delta = now() - due
if cur_dt > due:
return humanize.naturaldelta(
delta
) + ' ago'
else:
return 'in ' + humanize.naturaldelta(
delta
)
@dataclass(frozen=True)
class TaskForUser:
id: int
name: str
description: str
due: datetime|None
due_rel: str
created: datetime
complete: bool
completed: datetime|None
overdue: bool
namespace_id: int
category: str|None
tags: list[str]
mental_burn: int
social_burn: int
time_estimate: int|None
time_spent: int
importance: int
can_edit: bool
can_delete: bool
can_complete: bool
can_uncomplete: bool
@property
def delay_style(self) -> str:
''' Returns an appropriate 'animation-delay' property for
recently-completed tasks; otherwise an empty string.
Intended for use in jinja templating
'''
if self.just_complete and self.completed is not None:
delta = now() - self.completed
return f'animation-delay: {-1*delta.seconds}s;'
else:
return ''
@property
def just_complete(self) -> bool:
''' Returns whether the task was recently finished (within 5 minutes).
Used for jinja templates.
'''
return (
self.complete
and (self.completed is not None)
and (
(now() - self.completed)
< timedelta(minutes=5)
)
)
@classmethod
def from_task(
cls,
tsk: Task,
usr: User
) -> Either[Exception, Self]:
''' Returns a user-sanitized version of the task.
This function performs authorization checks on ``usr`` and will
return ``Left(AuthorizationError)`` if the action is unauthorized.
However, it is the caller's responsibility to report the violation
if warranted (use ``taskflower.auth.violations.check_for_auth_error_and_report``
in an ``lside_effect()``)
'''
perms = get_user_perms_on_task(usr, tsk)
cat = (
db_fetch_by_id(
TaskCategory,
tsk.category,
db
).flat_map(
lambda ct: Either.do_assert(
ct.namespace == tsk.namespace
).map(lambda _: ct)
).and_then(
lambda cat: cat.name,
lambda exc: None
)
if tsk.category is not None
else None
)
tags = gather_successes([
db_fetch_by_id(
Tag,
t2t.tag,
db
).flat_map(
lambda tag: Either.do_assert(
tag.namespace == tsk.namespace
).map(lambda _: tag)
).map(
lambda tag: tag.name
)
for t2t in db.session.query(
TagToTask
).filter(
TagToTask.task == tsk.id
).all()
])
return assert_user_perms_on_task(
usr,
tsk,
NamespacePermissionType.READ,
'Retrieve task'
).map(
lambda val: cls(
tsk.id,
tsk.name,
tsk.description,
tsk.due,
_due_str(tsk.due),
tsk.created,
tsk.complete,
tsk.completed,
(tsk.due < now()) if tsk.due else False,
tsk.namespace,
cat,
tags,
tsk.mental_burn,
tsk.social_burn,
tsk.time_estimate,
tsk.time_spent,
tsk.importance,
NPT.EDIT_ALL_TASKS in perms,
NPT.DELETE_ALL_TASKS in perms,
NPT.COMPLETE_ALL_TASKS in perms,
NPT.UNCOMPLETE_ALL_TASKS in perms
)
)