FilesExpand file tree

 main

/

task_services.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

82 lines (60 loc) · 2.33 KB

 main

/

task_services.py

Copy path

Top

File metadata and controls

• Code
• Blame

82 lines (60 loc) · 2.33 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

from models.task import Task

from schemas.task_schemas import TaskCreate, TaskUpdate

from models.user import User

from tortoise.expressions import Q

# Service class for managing tasks with many to many relationship with users

class TaskService:

@staticmethod

async def create_task(data: TaskCreate, owner_id: int):

task = await Task.create(

title=data.title,

description=data.description,

owner_id=owner_id

)

# Assign task to users if provided

if hasattr(data, 'assigned_to_ids') and data.assigned_to_ids:

users = await User.filter(id__in=data.assigned_to_ids).all()

if len(users) != len(data.assigned_to_ids):

return None

await task.assigned_to.add(*users)

return task

@staticmethod

async def get_user_task(user_id:int):

tasks = await Task.filter(

(Q(assigned_to__id=user_id) | Q(owner_id=user_id))

).distinct()

return tasks

@staticmethod

async def get_task_by_id(task_id:int,user_id:int):

task = await Task.filter(

Q(id=task_id) &

(Q(assigned_to__id=user_id) | Q(owner_id=user_id))

).first()

return task

#only owner can update the task

@staticmethod

async def update_task(task_id:int,data:TaskUpdate,owner_id:int):

task = await Task.filter(id=task_id,owner_id=owner_id).first()

if not task:

return None

if data.title is not None:

task.title = data.title

if data.description is not None:

task.description = data.description

if data.completed is not None:

task.completed = data.completed

if data.assigned_to is not None:

users = await User.filter(id__in=data.assigned_to).all()

if len(users) != len(data.assigned_to):

return None

await task.assigned_to.clear()

await task.assigned_to.add(*users)

await task.save()

return task

@staticmethod

async def delete_task(task_id:int,owner_id:int):

task = await Task.filter(id=task_id,owner_id=owner_id).first()

if not task:

return False

await task.delete()

return True