сделал базовые сервисы для всех таблиц

This commit is contained in:
Андрей Дувакин 2025-02-09 15:30:10 +05:00
parent 6079d975d3
commit 9f96d15567
29 changed files with 1465 additions and 19 deletions

View File

@ -1,6 +1,6 @@
from sqlalchemy.orm import Session
from app.domain.models.lecture import Lecture
from app.domain.models.lectures import Lecture
class LecturesRepository:

View File

@ -16,17 +16,6 @@ class UsersRepository:
def get_by_login(self, login: str):
return self.db.query(User).filter(User.login == login).first()
def change_password(self, user_id: int, old_password: str, new_password: str):
user = self.db.query(User).filter(User.id == user_id).first()
if user.check_password(old_password):
user.set_password(new_password)
self.db.merge(user)
self.db.commit()
self.db.refresh(user)
return user
return None
def create(self, user: User):
self.db.add(user)
self.db.commit()

View File

@ -12,7 +12,7 @@ class NotificationEntity(BaseModel):
user_id: int
type_id: int
course_id: Optional[int]
course_id: Optional[int] = None
class Config:
from_attributes = True

View File

@ -8,8 +8,8 @@ class TaskAnswerEntity(BaseModel):
id: Optional[int] = None
answer_date: datetime.datetime
text: str
is_current: Optional[bool]
comment: Optional[str]
is_current: Optional[bool] = None
comment: Optional[str] = None
user_id: int
task_id: int

View File

@ -1,4 +1,4 @@
from datetime import datetime
import datetime
from typing import Optional
from pydantic import BaseModel

View File

@ -8,12 +8,12 @@ class UserEntity(BaseModel):
id: Optional[int]
first_name: str
last_name: str
patronymic: Optional[str]
patronymic: Optional[str] = None
gender: str
birthday: datetime.date
registration_date: datetime.datetime
login: str
password: str
password: Optional[str] = None
email: str
role_id: int

View File

@ -9,7 +9,7 @@ import app.domain.models.appeals_topics
import app.domain.models.categories
import app.domain.models.courses
import app.domain.models.course_students
import app.domain.models.lecture
import app.domain.models.lectures
import app.domain.models.lessons
import app.domain.models.notification_types
import app.domain.models.notifications

View File

@ -0,0 +1,68 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.answer_files_repository import AnswerFilesRepository
from app.domain.entities.answer_files_entitity import AnswerFileEntity
from app.domain.models.answer_files import AnswerFile
class AnswerFilesService:
def __init__(self, db: Session):
self.answer_files_repository = AnswerFilesRepository(db)
def get_all(self) -> list[AnswerFileEntity]:
files = self.answer_files_repository.get_all()
return [
AnswerFileEntity(
id=file.id,
answer_id=file.answer_id,
file_title=file.file_title,
file_path=file.file_path
) for file in files
]
def get_by_id(self, answer_file_id: int) -> Optional[AnswerFileEntity]:
file = self.answer_files_repository.get_by_id(answer_file_id)
if file:
return AnswerFileEntity(
id=file.id,
answer_id=file.answer_id,
file_title=file.file_title,
file_path=file.file_path
)
return None
def create(self, answer_file: AnswerFileEntity) -> AnswerFileEntity:
file_model = AnswerFile(
answer_id=answer_file.answer_id,
file_title=answer_file.file_title,
file_path=answer_file.file_path
)
created_file = self.answer_files_repository.create(file_model)
return AnswerFileEntity(
id=created_file.id,
answer_id=created_file.answer_id,
file_title=created_file.file_title,
file_path=created_file.file_path
)
def update(self, answer_file_id: int, answer_file: AnswerFileEntity) -> Optional[AnswerFileEntity]:
file_model = self.answer_files_repository.get_by_id(answer_file_id)
if file_model:
file_model.answer_id = answer_file.answer_id
file_model.file_title = answer_file.file_title
file_model.file_path = answer_file.file_path
updated_file = self.answer_files_repository.update(file_model)
return AnswerFileEntity(
id=updated_file.id,
answer_id=updated_file.answer_id,
file_title=updated_file.file_title,
file_path=updated_file.file_path
)
return None
def delete(self, answer_file_id: int) -> bool:
return self.answer_files_repository.delete(answer_file_id) is not None

View File

@ -0,0 +1,68 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.answer_options_repository import AnswerOptionsRepository
from app.domain.entities.answer_options_entity import AnswerOptionEntity
from app.domain.models.answer_options import AnswerOption
class AnswerOptionsService:
def __init__(self, db: Session):
self.answer_options_repository = AnswerOptionsRepository(db)
def get_all(self) -> list[AnswerOptionEntity]:
options = self.answer_options_repository.get_all()
return [
AnswerOptionEntity(
id=option.id,
task_id=option.task_id,
text=option.text,
is_correct=option.is_correct,
) for option in options
]
def get_by_id(self, answer_option_id: int) -> Optional[AnswerOptionEntity]:
option = self.answer_options_repository.get_by_id(answer_option_id)
if option:
return AnswerOptionEntity(
id=option.id,
task_id=option.task_id,
text=option.text,
is_correct=option.is_correct,
)
return None
def create(self, answer_option: AnswerOptionEntity) -> AnswerOptionEntity:
option_model = AnswerOption(
task_id=answer_option.task_id,
text=answer_option.text,
is_correct=answer_option.is_correct,
)
created_option = self.answer_options_repository.create(option_model)
return AnswerOptionEntity(
id=created_option.id,
task_id=created_option.task_id,
text=created_option.text,
is_correct=created_option.is_correct,
)
def update(self, answer_option_id: int, answer_option: AnswerOptionEntity) -> Optional[AnswerOptionEntity]:
option_model = self.answer_options_repository.get_by_id(answer_option_id)
if option_model:
option_model.task_id = answer_option.task_id
option_model.text = answer_option.text
option_model.is_correct = answer_option.is_correct
updated_option = self.answer_options_repository.update(option_model)
return AnswerOptionEntity(
id=updated_option.id,
task_id=updated_option.task_id,
text=updated_option.text,
is_correct=updated_option.is_correct,
)
return None
def delete(self, answer_option_id: int) -> bool:
return self.answer_options_repository.delete(answer_option_id) is not None

View File

@ -0,0 +1,80 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.appeals_repository import AppealsRepository
from app.domain.entities.appeals_entity import AppealEntity
from app.domain.models.appeals import Appeal
class AppealsService:
def __init__(self, db: Session):
self.appeals_repository = AppealsRepository(db)
def get_all(self) -> list[AppealEntity]:
appeals = self.appeals_repository.get_all()
return [
AppealEntity(
id=appeal.id,
message=appeal.message,
created_date=appeal.created_date,
status=appeal.status,
topic_id=appeal.topic_id,
user_id=appeal.user_id,
) for appeal in appeals
]
def get_by_id(self, appeal_id: int) -> Optional[AppealEntity]:
appeal = self.appeals_repository.get_by_id(appeal_id)
if appeal:
return AppealEntity(
id=appeal.id,
message=appeal.message,
created_date=appeal.created_date,
status=appeal.status,
topic_id=appeal.topic_id,
user_id=appeal.user_id,
)
return None
def create(self, appeal: AppealEntity) -> AppealEntity:
appeal_model = Appeal(
message=appeal.message,
created_date=appeal.created_date,
status=appeal.status,
topic_id=appeal.topic_id,
user_id=appeal.user_id,
)
created_appeal = self.appeals_repository.create(appeal_model)
return AppealEntity(
id=created_appeal.id,
message=created_appeal.message,
created_date=created_appeal.created_date,
status=created_appeal.status,
topic_id=created_appeal.topic_id,
user_id=created_appeal.user_id,
)
def update(self, appeal_id: int, appeal: AppealEntity) -> Optional[AppealEntity]:
appeal_model = self.appeals_repository.get_by_id(appeal_id)
if appeal_model:
appeal_model.message = appeal.message
appeal_model.created_date = appeal.created_date
appeal_model.status = appeal.status
appeal_model.topic_id = appeal.topic_id
appeal_model.user_id = appeal.user_id
updated_appeal = self.appeals_repository.update(appeal_model)
return AppealEntity(
id=updated_appeal.id,
message=updated_appeal.message,
created_date=updated_appeal.created_date,
status=updated_appeal.status,
topic_id=updated_appeal.topic_id,
user_id=updated_appeal.user_id,
)
return None
def delete(self, appeal_id: int) -> bool:
return self.appeals_repository.delete(appeal_id) is not None

View File

@ -0,0 +1,56 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.appeals_topics_repository import AppealsTopicsRepository
from app.domain.entities.appeals_topics_entity import AppealTopicEntity
from app.domain.models.appeals_topics import AppealsTopic
class AppealsTopicService:
def __init__(self, db: Session):
self.appeals_topics_repository = AppealsTopicsRepository(db)
def get_all(self) -> list[AppealTopicEntity]:
appeals_topics = self.appeals_topics_repository.get_all()
return [
AppealTopicEntity(
id=appeals_topic.id,
title=appeals_topic.title,
) for appeals_topic in appeals_topics
]
def get_by_id(self, appeals_topic_id: int) -> Optional[AppealTopicEntity]:
appeals_topic = self.appeals_topics_repository.get_by_id(appeals_topic_id)
if appeals_topic:
return AppealTopicEntity(
id=appeals_topic.id,
title=appeals_topic.title,
)
return None
def create(self, appeals_topic: AppealTopicEntity) -> AppealTopicEntity:
appeals_model = AppealsTopic(
title=appeals_topic.title,
)
created_appeals_topic = self.appeals_topics_repository.create(appeals_model)
return AppealTopicEntity(
id=created_appeals_topic.id,
title=created_appeals_topic.title,
)
def update(self, appeals_topic_id: int, appeals_topic: AppealTopicEntity) -> Optional[AppealTopicEntity]:
appeals_topic_model = self.appeals_topics_repository.get_by_id(appeals_topic_id)
if appeals_topic_model:
appeals_topic_model.title = appeals_topic.title
updated_appeals_topic = self.appeals_topics_repository.update(appeals_topic_model)
return AppealTopicEntity(
id=updated_appeals_topic.id,
title=updated_appeals_topic.title,
)
return None
def delete(self, appeals_topic_id: int) -> bool:
return self.appeals_topics_repository.delete(appeals_topic_id) is not None

View File

@ -0,0 +1,62 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.categories_repository import CategoriesRepository
from app.domain.entities.categories_entity import CategoryEntity
from app.domain.models.categories import Category
class CategoriesService:
def __init__(self, db: Session):
self.categories_repository = CategoriesRepository(db)
def get_all(self) -> list[CategoryEntity]:
categories = self.categories_repository.get_all()
return [
CategoryEntity(
id=category.id,
title=category.title,
description=category.description,
) for category in categories
]
def get_by_id(self, category_id: int) -> Optional[CategoryEntity]:
category = self.categories_repository.get_by_id(category_id)
if category:
return CategoryEntity(
id=category.id,
title=category.title,
description=category.description,
)
return None
def create(self, category: CategoryEntity) -> CategoryEntity:
category_model = Category(
title=category.title,
description=category.description,
)
created_category = self.categories_repository.create(category_model)
return CategoryEntity(
id=created_category.id,
title=created_category.title,
description=created_category.description,
)
def update(self, category_id: int, category: CategoryEntity) -> Optional[CategoryEntity]:
category_model = self.categories_repository.get_by_id(category_id)
if category_model:
category_model.title = category.title
category_model.description = category.description
updated_category = self.categories_repository.update(category_model)
return CategoryEntity(
id=updated_category.id,
title=updated_category.title,
description=updated_category.description,
)
return None
def delete(self, category_id: int) -> bool:
return self.categories_repository.delete(category_id) is not None

View File

@ -0,0 +1,74 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.course_students_repository import CourseStudentsRepository
from app.domain.entities.course_students_entity import CourseStudentEntity
from app.domain.models.course_students import CourseStudent
class CourseStudentsService:
def __init__(self, db: Session):
self.course_students_repository = CourseStudentsRepository(db)
def get_all(self) -> list[CourseStudentEntity]:
course_students = self.course_students_repository.get_all()
return [
CourseStudentEntity(
id=course_student.id,
enrollment_date=course_student.enrollment_date,
is_finished=course_student.is_finished,
course_id=course_student.course_id,
user_id=course_student.user_id,
) for course_student in course_students
]
def get_by_id(self, course_student_id: int) -> Optional[CourseStudentEntity]:
course_student = self.course_students_repository.get_by_id(course_student_id)
if course_student:
return CourseStudentEntity(
id=course_student.id,
enrollment_date=course_student.enrollment_date,
is_finished=course_student.is_finished,
course_id=course_student.course_id,
user_id=course_student.user_id,
)
return None
def create(self, course_student: CourseStudentEntity) -> CourseStudentEntity:
course_student_model = CourseStudent(
enrollment_date=course_student.enrollment_date,
is_finished=course_student.is_finished,
course_id=course_student.course_id,
user_id=course_student.user_id,
)
created_course_student = self.course_students_repository.create(course_student_model)
return CourseStudentEntity(
id=created_course_student.id,
enrollment_date=created_course_student.enrollment_date,
is_finished=created_course_student.is_finished,
course_id=created_course_student.course_id,
user_id=created_course_student.user_id,
)
def update(self, course_student_id: int, course_student: CourseStudentEntity) -> Optional[CourseStudentEntity]:
course_student_model = self.course_students_repository.get_by_id(course_student_id)
if course_student_model:
course_student_model.enrollment_date = course_student.enrollment_date
course_student_model.is_finished = course_student.is_finished
course_student_model.course_id = course_student.course_id
course_student_model.user_id = course_student.user_id
updated_course_student = self.course_students_repository.update(course_student_model)
return CourseStudentEntity(
id=updated_course_student.id,
enrollment_date=updated_course_student.enrollment_date,
is_finished=updated_course_student.is_finished,
course_id=updated_course_student.course_id,
user_id=updated_course_student.user_id,
)
return None
def delete(self, course_student_id: int) -> bool:
return self.course_students_repository.delete(course_student_id) is not None

View File

@ -0,0 +1,74 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.courses_repository import CoursesRepository
from app.domain.entities.courses_entity import CourseEntity
from app.domain.models.courses import Course
class CoursesService:
def __init__(self, db: Session):
self.courses_repository = CoursesRepository(db)
def get_all(self) -> list[CourseEntity]:
courses = self.courses_repository.get_all()
return [
CourseEntity(
id=course.id,
title=course.title,
description=course.description,
category_id=course.category_id,
owner_user_id=course.owner_user_id,
) for course in courses
]
def get_by_id(self, course_id: int) -> Optional[CourseEntity]:
course = self.courses_repository.get_by_id(course_id)
if course:
return CourseEntity(
id=course.id,
title=course.title,
description=course.description,
category_id=course.category_id,
owner_user_id=course.owner_user_id,
)
return None
def create(self, course: CourseEntity) -> CourseEntity:
course_model = Course(
title=course.title,
description=course.description,
category_id=course.category_id,
owner_user_id=course.owner_user_id,
)
created_course = self.courses_repository.create(course_model)
return CourseEntity(
id=created_course.id,
title=created_course.title,
description=created_course.description,
category_id=created_course.category_id,
owner_user_id=created_course.owner_user_id,
)
def update(self, course_id: int, course: CourseEntity) -> Optional[CourseEntity]:
course_model = self.courses_repository.get_by_id(course_id)
if course_model:
course_model.title = course.title
course_model.description = course.description
course_model.category_id = course.category_id
course_model.owner_user_id = course.owner_user_id
updated_course = self.courses_repository.update(course_model)
return CourseEntity(
id=updated_course.id,
title=updated_course.title,
description=updated_course.description,
category_id=updated_course.category_id,
owner_user_id=updated_course.owner_user_id,
)
return None
def delete(self, course_id: int) -> bool:
return self.courses_repository.delete(course_id) is not None

View File

@ -0,0 +1,74 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.lectures_repository import LecturesRepository
from app.domain.entities.lectures_entity import LectureEntity
from app.domain.models.lectures import Lecture
class LecturesService:
def __init__(self, db: Session):
self.lectures_repository = LecturesRepository(db)
def get_all(self) -> list[LectureEntity]:
lectures = self.lectures_repository.get_all()
return [
LectureEntity(
id=lecture.id,
text=lecture.text,
image=lecture.image,
number=lecture.number,
step_id=lecture.step_id
) for lecture in lectures
]
def get_by_id(self, lecture_id: int) -> Optional[LectureEntity]:
lecture = self.lectures_repository.get_by_id(lecture_id)
if lecture:
return LectureEntity(
id=lecture.id,
text=lecture.text,
image=lecture.image,
number=lecture.number,
step_id=lecture.step_id
)
return None
def create(self, lecture: LectureEntity) -> LectureEntity:
lecture_model = Lecture(
text=lecture.text,
image=lecture.image,
number=lecture.number,
step_id=lecture.step_id
)
created_lecture = self.lectures_repository.create(lecture_model)
return LectureEntity(
id=created_lecture.id,
text=created_lecture.text,
image=created_lecture.image,
number=created_lecture.number,
step_id=created_lecture.step_id
)
def update(self, lecture_id: int, lecture: LectureEntity) -> Optional[LectureEntity]:
lecture_model = self.lectures_repository.get_by_id(lecture_id)
if lecture_model:
lecture_model.text = lecture.text
lecture_model.image = lecture.image
lecture_model.number = lecture.number
lecture_model.step_id = lecture.step_id
updated_lecture = self.lectures_repository.update(lecture_model)
return LectureEntity(
id=updated_lecture.id,
text=updated_lecture.text,
image=updated_lecture.image,
number=updated_lecture.number,
step_id=updated_lecture.step_id
)
return None
def delete(self, lecture_id: int) -> bool:
return self.lectures_repository.delete(lecture_id) is not None

View File

@ -0,0 +1,68 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.lessons_repository import LessonsRepository
from app.domain.entities.lessons_entity import LessonEntity
from app.domain.models.lessons import Lesson
class LessonsService:
def __init__(self, db: Session):
self.lessons_repository = LessonsRepository(db)
def get_all(self) -> list[LessonEntity]:
lessons = self.lessons_repository.get_all()
return [
LessonEntity(
id=lesson.id,
title=lesson.title,
description=lesson.description,
course_id=lesson.course_id,
) for lesson in lessons
]
def get_by_id(self, lesson_id: int) -> Optional[LessonEntity]:
lesson = self.lessons_repository.get_by_id(lesson_id)
if lesson:
return LessonEntity(
id=lesson.id,
title=lesson.title,
description=lesson.description,
course_id=lesson.course_id,
)
return None
def create(self, lesson: LessonEntity) -> LessonEntity:
lesson_model = Lesson(
title=lesson.title,
description=lesson.description,
course_id=lesson.course_id,
)
created_lesson = self.lessons_repository.create(lesson_model)
return LessonEntity(
id=created_lesson.id,
title=created_lesson.title,
description=created_lesson.description,
course_id=created_lesson.course_id,
)
def update(self, lesson_id: int, lesson: LessonEntity) -> Optional[LessonEntity]:
lesson_model = self.lessons_repository.get_by_id(lesson_id)
if lesson_model:
lesson_model.title = lesson.title
lesson_model.description = lesson.description
lesson_model.course_id = lesson.course_id
updated_lesson = self.lessons_repository.update(lesson_model)
return LessonEntity(
id=updated_lesson.id,
title=updated_lesson.title,
description=updated_lesson.description,
course_id=updated_lesson.course_id,
)
return None
def delete(self, lesson_id: int) -> bool:
return self.lessons_repository.delete(lesson_id) is not None

View File

@ -0,0 +1,63 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.notification_types_repository import NotificationTypesRepository
from app.domain.entities.notification_types_entity import NotificationTypeEntity
from app.domain.models.notification_types import NotificationType
class NotificationTypesService:
def __init__(self, db: Session):
self.notification_types_repository = NotificationTypesRepository(db)
def get_all(self) -> list[NotificationTypeEntity]:
notification_types = self.notification_types_repository.get_all()
return [
NotificationTypeEntity(
id=notification_type.id,
title=notification_type.title,
description=notification_type.description,
) for notification_type in notification_types
]
def get_by_id(self, notification_type_id: int) -> Optional[NotificationTypeEntity]:
notification_type = self.notification_types_repository.get_by_id(notification_type_id)
if notification_type:
return NotificationTypeEntity(
id=notification_type.id,
title=notification_type.title,
description=notification_type.description,
)
return None
def create(self, notification_type: NotificationTypeEntity) -> NotificationTypeEntity:
notification_type_model = NotificationType(
title=notification_type.title,
description=notification_type.description,
)
created_notification_type = self.notification_types_repository.create(notification_type_model)
return NotificationTypeEntity(
id=created_notification_type.id,
title=created_notification_type.title,
description=created_notification_type.description,
)
def update(self, notification_type_id: int, notification_type: NotificationTypeEntity) -> Optional[
NotificationTypeEntity]:
notification_type_model = self.notification_types_repository.get_by_id(notification_type_id)
if notification_type_model:
notification_type_model.title = notification_type.title
notification_type_model.description = notification_type.description
updated_notification_type = self.notification_types_repository.update(notification_type_model)
return NotificationTypeEntity(
id=updated_notification_type.id,
title=updated_notification_type.title,
description=updated_notification_type.description,
)
return None
def delete(self, notification_type_id: int) -> bool:
return self.notification_types_repository.delete(notification_type_id) is not None

View File

@ -0,0 +1,85 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.notifications_repository import NotificationsRepository
from app.domain.entities.notification_entity import NotificationEntity
class NotificationsService:
def __init__(self, db: Session):
self.notifications_repository = NotificationsRepository(db)
def get_all(self) -> list[NotificationEntity]:
notifications = self.notifications_repository.get_all()
return [
NotificationEntity(
id=notification.id,
text=notification.text,
datetime_notification=notification.datetime_notification,
is_read=notification.is_read,
user_id=notification.user_id,
type_id=notification.type_id,
course_id=notification.course_id,
) for notification in notifications
]
def get_by_id(self, notification_id: int) -> Optional[NotificationEntity]:
notification = self.notifications_repository.get_by_id(notification_id)
if notification:
return NotificationEntity(
id=notification.id,
text=notification.text,
datetime_notification=notification.datetime_notification,
is_read=notification.is_read,
user_id=notification.user_id,
type_id=notification.type_id,
course_id=notification.course_id,
)
return None
def create(self, notification: NotificationEntity) -> NotificationEntity:
notification_model = Notification(
text=notification.text,
datetime_notification=notification.datetime_notification,
is_read=notification.is_read,
user_id=notification.user_id,
type_id=notification.type_id,
course_id=notification.course_id,
)
created_notification = self.notifications_repository.create(notification_model)
return NotificationEntity(
id=created_notification.id,
text=created_notification.text,
datetime_notification=created_notification.datetime_notification,
is_read=created_notification.is_read,
user_id=created_notification.user_id,
type_id=created_notification.type_id,
course_id=created_notification.course_id,
)
def update(self, notification_id: int, notification: NotificationEntity) -> Optional[NotificationEntity]:
notification_model = self.notifications_repository.get_by_id(notification_id)
if notification_model:
notification_model.text = notification.text
notification_model.datetime_notification = notification.datetime_notification
notification_model.is_read = notification.is_read
notification_model.user_id = notification.user_id
notification_model.type_id = notification.type_id
notification_model.course_id = notification.course_id
updated_notification = self.notifications_repository.update(notification_model)
return NotificationEntity(
id=updated_notification.id,
text=updated_notification.text,
datetime_notification=updated_notification.datetime_notification,
is_read=updated_notification.is_read,
user_id=updated_notification.user_id,
type_id=updated_notification.type_id,
course_id=updated_notification.course_id,
)
return None
def delete(self, notification_id: int) -> bool:
return self.notifications_repository.delete(notification_id) is not None

View File

@ -0,0 +1,56 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.roles_repository import RolesRepository
from app.domain.entities.roles_entity import RoleEntity
from app.domain.models.roles import Role
class RolesService:
def __init__(self, db: Session):
self.roles_repository = RolesRepository(db)
def get_all(self) -> list[RoleEntity]:
roles = self.roles_repository.get_all()
return [
RoleEntity(
id=role.id,
title=role.title,
) for role in roles
]
def get_by_id(self, role_id: int) -> Optional[RoleEntity]:
role = self.roles_repository.get_by_id(role_id)
if role:
return RoleEntity(
id=role.id,
title=role.title,
)
return None
def create(self, role: RoleEntity) -> RoleEntity:
role_model = Role(
title=role.title,
)
created_role = self.roles_repository.create(role_model)
return RoleEntity(
id=created_role.id,
title=created_role.title,
)
def update(self, role_id: int, role: RoleEntity) -> Optional[RoleEntity]:
role_model = self.roles_repository.get_by_id(role_id)
if role_model:
role_model.title = role.title
updated_role = self.roles_repository.update(role_model)
return RoleEntity(
id=updated_role.id,
title=updated_role.title,
)
return None
def delete(self, role_id: int) -> bool:
return self.roles_repository.delete(role_id) is not None

View File

@ -0,0 +1,68 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.step_tasks_repository import StepTasksRepository
from app.domain.entities.step_tasks_entity import StepTaskEntity
from app.domain.models.step_tasks import StepTask
class StepTasksService:
def __init__(self, db: Session):
self.step_tasks_repository = StepTasksRepository(db)
def get_all(self) -> list[StepTaskEntity]:
step_tasks = self.step_tasks_repository.get_all()
return [
StepTaskEntity(
id=step_task.id,
text=step_task.text,
step_id=step_task.step_id,
type_id=step_task.type_id,
) for step_task in step_tasks
]
def get_by_id(self, step_task_id: int) -> Optional[StepTaskEntity]:
step_task = self.step_tasks_repository.get_by_id(step_task_id)
if step_task:
return StepTaskEntity(
id=step_task.id,
text=step_task.text,
step_id=step_task.step_id,
type_id=step_task.type_id,
)
return None
def create(self, step_task: StepTaskEntity) -> StepTaskEntity:
step_task_model = StepTask(
text=step_task.text,
step_id=step_task.step_id,
type_id=step_task.type_id,
)
created_step_task = self.step_tasks_repository.create(step_task_model)
return StepTaskEntity(
id=created_step_task.id,
text=created_step_task.text,
step_id=created_step_task.step_id,
type_id=created_step_task.type_id,
)
def update(self, step_task_id: int, step_task: StepTaskEntity) -> Optional[StepTaskEntity]:
step_task_model = self.step_tasks_repository.get_by_id(step_task_id)
if step_task_model:
step_task_model.text = step_task.text
step_task_model.step_id = step_task.step_id
step_task_model.type_id = step_task.type_id
updated_step_task = self.step_tasks_repository.update(step_task_model)
return StepTaskEntity(
id=updated_step_task.id,
text=updated_step_task.text,
step_id=updated_step_task.step_id,
type_id=updated_step_task.type_id,
)
return None
def delete(self, step_task_id: int) -> bool:
return self.step_tasks_repository.delete(step_task_id) is not None

View File

@ -0,0 +1,62 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.step_types_repository import StepTypesRepository
from app.domain.entities.step_types_entity import StepTypeEntity
from app.domain.models.step_types import StepType
class StepTypesService:
def __init__(self, db: Session):
self.step_types_repository = StepTypesRepository(db)
def get_all(self) -> list[StepTypeEntity]:
step_types = self.step_types_repository.get_all()
return [
StepTypeEntity(
id=step_type.id,
title=step_type.title,
description=step_type.description,
) for step_type in step_types
]
def get_by_id(self, step_type_id: int) -> Optional[StepTypeEntity]:
step_type = self.step_types_repository.get_by_id(step_type_id)
if step_type:
return StepTypeEntity(
id=step_type.id,
title=step_type.title,
description=step_type.description,
)
return None
def create(self, step_type: StepTypeEntity) -> StepTypeEntity:
step_type_model = StepType(
title=step_type.title,
description=step_type.description,
)
created_step_type = self.step_types_repository.create(step_type_model)
return StepTypeEntity(
id=created_step_type.id,
title=created_step_type.title,
description=created_step_type.description,
)
def update(self, step_type_id: int, step_type: StepTypeEntity) -> Optional[StepTypeEntity]:
step_type_model = self.step_types_repository.get_by_id(step_type_id)
if step_type_model:
step_type_model.title = step_type.title
step_type_model.description = step_type.description
updated_step_type = self.step_types_repository.update(step_type_model)
return StepTypeEntity(
id=updated_step_type.id,
title=updated_step_type.title,
description=updated_step_type.description,
)
return None
def delete(self, step_type_id: int) -> bool:
return self.step_types_repository.delete(step_type_id) is not None

View File

@ -0,0 +1,69 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.steps_repository import StepsRepository
from app.domain.entities.steps_entity import StepEntity
from app.domain.models.steps import Step
class StepsService:
def __init__(self, db: Session):
self.steps_repository = StepsRepository(db)
def get_all(self) -> list[StepEntity]:
steps = self.steps_repository.get_all()
return [
StepEntity(
id=step.id,
title=step.title,
lesson_id=step.lesson_id,
type_id=step.type_id
)
for step in steps
]
def get_by_id(self, step_id: int) -> Optional[StepEntity]:
step = self.steps_repository.get_by_id(step_id)
if step:
return StepEntity(
id=step.id,
title=step.title,
lesson_id=step.lesson_id,
type_id=step.type_id
)
return None
def create(self, step: StepEntity) -> StepEntity:
step_model = Step(
title=step.title,
lesson_id=step.lesson_id,
type_id=step.type_id
)
created_step = self.steps_repository.create(step_model)
return StepEntity(
id=created_step.id,
title=created_step.title,
lesson_id=created_step.lesson_id,
type_id=created_step.type_id
)
def update(self, step_id: int, step: StepEntity) -> Optional[StepEntity]:
step_model = self.steps_repository.get_by_id(step_id)
if step_model:
step_model.title = step.title
step_model.lesson_id = step.lesson_id
step_model.type_id = step.type_id
updated_step = self.steps_repository.update(step_model)
return StepEntity(
id=updated_step.id,
title=updated_step.title,
lesson_id=updated_step.lesson_id,
type_id=updated_step.type_id
)
return None
def delete(self, step_id: int) -> bool:
return self.steps_repository.delete(step_id) is not None

View File

@ -0,0 +1,86 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.task_answers_repository import TaskAnswersRepository
from app.domain.entities.task_answers_entity import TaskAnswerEntity
from app.domain.models.task_answers import TaskAnswer
class TaskAnswersService:
def __init__(self, db: Session):
self.task_answers_repository = TaskAnswersRepository(db)
def get_all(self) -> list[TaskAnswerEntity]:
task_answers = self.task_answers_repository.get_all()
return [
TaskAnswerEntity(
id=task_answer.id,
answer_date=task_answer.answer_date,
text=task_answer.text,
is_current=task_answer.is_current,
comment=task_answer.comment,
user_id=task_answer.user_id,
task_id=task_answer.task_id,
) for task_answer in task_answers
]
def get_by_id(self, task_answer_id: int) -> Optional[TaskAnswerEntity]:
task_answer = self.task_answers_repository.get_by_id(task_answer_id)
if task_answer:
return TaskAnswerEntity(
id=task_answer.id,
answer_date=task_answer.answer_date,
text=task_answer.text,
is_current=task_answer.is_current,
comment=task_answer.comment,
user_id=task_answer.user_id,
task_id=task_answer.task_id,
)
return None
def create(self, task_answer: TaskAnswerEntity) -> TaskAnswerEntity:
task_answer_model = TaskAnswer(
answer_date=task_answer.answer_date,
text=task_answer.text,
is_current=task_answer.is_current,
comment=task_answer.comment,
user_id=task_answer.user_id,
task_id=task_answer.task_id,
)
created_task_answer = self.task_answers_repository.create(task_answer_model)
return TaskAnswerEntity(
id=created_task_answer.id,
answer_date=created_task_answer.answer_date,
text=created_task_answer.text,
is_current=created_task_answer.is_current,
comment=created_task_answer.comment,
user_id=created_task_answer.user_id,
task_id=created_task_answer.task_id,
)
def update(self, task_answer_id: int, task_answer: TaskAnswerEntity) -> Optional[TaskAnswerEntity]:
task_answer_model = self.task_answers_repository.get_by_id(task_answer_id)
if task_answer_model:
task_answer_model.answer_date = task_answer.answer_date
task_answer_model.text = task_answer.text
task_answer_model.is_current = task_answer.is_current
task_answer_model.comment = task_answer.comment
task_answer_model.user_id = task_answer.user_id
task_answer_model.task_id = task_answer.task_id
updated_task_answer = self.task_answers_repository.update(task_answer_model)
return TaskAnswerEntity(
id=updated_task_answer.id,
answer_date=updated_task_answer.answer_date,
text=updated_task_answer.text,
is_current=updated_task_answer.is_current,
comment=updated_task_answer.comment,
user_id=updated_task_answer.user_id,
task_id=updated_task_answer.task_id,
)
return None
def delete(self, task_answer_id: int) -> bool:
return self.task_answers_repository.delete(task_answer_id) is not None

View File

@ -0,0 +1,69 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.task_files_repository import TaskFilesRepository
from app.domain.entities.task_files_entity import TaskFileEntity
from app.domain.models.task_files import TaskFile
class TaskFilesService:
def __init__(self, db: Session):
self.task_files_repository = TaskFilesRepository(db)
def get_all(self) -> list[TaskFileEntity]:
task_files = self.task_files_repository.get_all()
return [
TaskFileEntity(
id=task_file.id,
file_path=task_file.file_path,
file_title=task_file.file_title,
task_id=task_file.task_id
)
for task_file in task_files
]
def get_by_id(self, task_file_id: int) -> Optional[TaskFileEntity]:
task_file = self.task_files_repository.get_by_id(task_file_id)
if task_file:
return TaskFileEntity(
id=task_file.id,
file_path=task_file.file_path,
file_title=task_file.file_title,
task_id=task_file.task_id
)
return None
def create(self, task_file: TaskFileEntity) -> TaskFileEntity:
task_file_model = TaskFile(
file_path=task_file.file_path,
file_title=task_file.file_title,
task_id=task_file.task_id
)
created_task_file = self.task_files_repository.create(task_file_model)
return TaskFileEntity(
id=created_task_file.id,
file_path=created_task_file.file_path,
file_title=created_task_file.file_title,
task_id=created_task_file.task_id
)
def update(self, task_file_id: int, task_file: TaskFileEntity) -> Optional[TaskFileEntity]:
task_file_model = self.task_files_repository.get_by_id(task_file_id)
if task_file_model:
task_file_model.file_path = task_file.file_path
task_file_model.file_title = task_file.file_title
task_file_model.task_id = task_file.task_id
updated_task_file = self.task_files_repository.update(task_file_model)
return TaskFileEntity(
id=updated_task_file.id,
file_path=updated_task_file.file_path,
file_title=updated_task_file.file_title,
task_id=updated_task_file.task_id
)
return None
def delete(self, task_file_id: int) -> bool:
return self.task_files_repository.delete(task_file_id) is not None

View File

@ -0,0 +1,63 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.task_types_repository import TaskTypesRepository
from app.domain.entities.task_types_entity import TaskTypeEntity
from app.domain.models.task_types import TaskType
class TaskTypesService:
def __init__(self, db: Session):
self.task_types_repository = TaskTypesRepository(db)
def get_all(self) -> list[TaskTypeEntity]:
task_types = self.task_types_repository.get_all()
return [
TaskTypeEntity(
id=task_type.id,
title=task_type.title,
description=task_type.description
)
for task_type in task_types
]
def get_by_id(self, task_type_id: int) -> Optional[TaskTypeEntity]:
task_type = self.task_types_repository.get_by_id(task_type_id)
if task_type:
return TaskTypeEntity(
id=task_type.id,
title=task_type.title,
description=task_type.description
)
return None
def create(self, task_type: TaskTypeEntity) -> TaskTypeEntity:
task_type_model = TaskType(
title=task_type.title,
description=task_type.description
)
created_task_type = self.task_types_repository.create(task_type_model)
return TaskTypeEntity(
id=created_task_type.id,
title=created_task_type.title,
description=created_task_type.description
)
def update(self, task_type_id: int, task_type: TaskTypeEntity) -> Optional[TaskTypeEntity]:
task_type_model = self.task_types_repository.get_by_id(task_type_id)
if task_type_model:
task_type_model.title = task_type.title
task_type_model.description = task_type.description
updated_task_type = self.task_types_repository.update(task_type_model)
return TaskTypeEntity(
id=updated_task_type.id,
title=updated_task_type.title,
description=updated_task_type.description
)
return None
def delete(self, task_type_id: int) -> None:
self.task_types_repository.delete(task_type_id)

View File

@ -0,0 +1,69 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.test_answers_repository import TestAnswersRepository
from app.domain.entities.test_answers_entity import TestAnswerEntity
from app.domain.models.test_answers import TestAnswer
class TestAnswersService:
def __init__(self, db: Session):
self.test_answers_repository = TestAnswersRepository(db)
def get_all(self) -> list[TestAnswerEntity]:
test_answers = self.test_answers_repository.get_all()
return [
TestAnswerEntity(
id=test_answer.id,
answer_date=test_answer.answer_date,
user_id=test_answer.user_id,
answer_id=test_answer.answer_id
)
for test_answer in test_answers
]
def get_by_id(self, test_answer_id: int) -> Optional[TestAnswerEntity]:
test_answer = self.test_answers_repository.get_by_id(test_answer_id)
if test_answer:
return TestAnswerEntity(
id=test_answer.id,
answer_date=test_answer.answer_date,
user_id=test_answer.user_id,
answer_id=test_answer.answer_id
)
return None
def create(self, test_answer: TestAnswerEntity) -> TestAnswerEntity:
test_answer_model = TestAnswer(
answer_date=test_answer.answer_date,
user_id=test_answer.user_id,
answer_id=test_answer.answer_id
)
created_test_answer = self.test_answers_repository.create(test_answer_model)
return TestAnswerEntity(
id=created_test_answer.id,
answer_date=created_test_answer.answer_date,
user_id=created_test_answer.user_id,
answer_id=created_test_answer.answer_id
)
def update(self, test_answer_id: int, test_answer: TestAnswerEntity) -> Optional[TestAnswerEntity]:
test_answer_model = self.test_answers_repository.get_by_id(test_answer_id)
if test_answer_model:
test_answer_model.answer_date = test_answer.answer_date
test_answer_model.user_id = test_answer.user_id
test_answer_model.answer_id = test_answer.answer_id
updated_test_answer = self.test_answers_repository.update(test_answer_model)
return TestAnswerEntity(
id=updated_test_answer.id,
answer_date=updated_test_answer.answer_date,
user_id=updated_test_answer.user_id,
answer_id=updated_test_answer.answer_id
)
return None
def delete(self, test_answer_id: int) -> None:
self.test_answers_repository.delete(test_answer_id)

View File

@ -0,0 +1,143 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.application.users_repository import UsersRepository
from app.domain.entities.users_entity import UserEntity
from app.domain.models.users import User
class UsersService:
def __init__(self, db: Session):
self.users_repository = UsersRepository(db)
def get_all(self) -> list[UserEntity]:
users = self.users_repository.get_all()
return [
UserEntity(
id=user.id,
first_name=user.first_name,
last_name=user.last_name,
patronymic=user.patronymic,
gender=user.gender,
birthday=user.birthday,
registration_date=user.registration_date,
login=user.login,
email=user.email,
role_id=user.role_id,
) for user in users
]
def get_by_id(self, user_id: int) -> Optional[UserEntity]:
user = self.users_repository.get_by_id(user_id)
if user:
return UserEntity(
id=user.id,
first_name=user.first_name,
last_name=user.last_name,
patronymic=user.patronymic,
gender=user.gender,
birthday=user.birthday,
registration_date=user.registration_date,
login=user.login,
email=user.email,
role_id=user.role_id,
)
return None
def get_by_login(self, login: str) -> Optional[UserEntity]:
user = self.users_repository.get_by_login(login)
if user:
return UserEntity(
id=user.id,
first_name=user.first_name,
last_name=user.last_name,
patronymic=user.patronymic,
gender=user.gender,
birthday=user.birthday,
registration_date=user.registration_date,
login=user.login,
email=user.email,
role_id=user.role_id,
)
return None
def create(self, user: UserEntity) -> UserEntity:
user_model = User(
first_name=user.first_name,
last_name=user.last_name,
patronymic=user.patronymic,
gender=user.gender,
birthday=user.birthday,
registration_date=user.registration_date,
login=user.login,
email=user.email,
role_id=user.role_id,
)
user_model.set_password(user.password)
created_user = self.users_repository.create(user_model)
return UserEntity(
id=created_user.id,
first_name=created_user.first_name,
last_name=created_user.last_name,
patronymic=created_user.patronymic,
gender=created_user.gender,
birthday=created_user.birthday,
registration_date=created_user.registration_date,
login=created_user.login,
email=created_user.email,
role_id=created_user.role_id,
)
def update(self, user: UserEntity) -> Optional[UserEntity]:
user_model = self.users_repository.get_by_id(user.id)
if user_model:
user_model.first_name = user.first_name
user_model.last_name = user.last_name
user_model.patronymic = user.patronymic
user_model.gender = user.gender
user_model.birthday = user.birthday
user_model.registration_date = user.registration_date
user_model.login = user.login
user_model.email = user.email
user_model.role_id = user.role_id
updated_user = self.users_repository.update(user_model)
return UserEntity(
id=updated_user.id,
first_name=updated_user.first_name,
last_name=updated_user.last_name,
patronymic=updated_user.patronymic,
gender=updated_user.gender,
birthday=updated_user.birthday,
registration_date=updated_user.registration_date,
login=updated_user.login,
email=updated_user.email,
role_id=updated_user.role_id,
)
return None
def update_password(self, user_id: int, old_password: str, new_password: str) -> Optional[UserEntity]:
user = self.users_repository.get_by_id(user_id)
if user and user.check_password(old_password):
user.set_password(new_password)
updated_user = self.users_repository.update(user)
return UserEntity(
id=updated_user.id,
first_name=updated_user.first_name,
last_name=updated_user.last_name,
patronymic=updated_user.patronymic,
gender=updated_user.gender,
birthday=updated_user.birthday,
registration_date=updated_user.registration_date,
login=updated_user.login,
email=updated_user.email,
role_id=updated_user.role_id,
)
return None
def delete(self, user_id: int) -> bool:
return self.users_repository.delete(user_id) is not None