Это вторая часть серии «Внедрение ООП-системы управления задачами». Предыдущая статья была посвящена общему дизайну приложения. Я создал пару диаграмм вместе с вариантами использования и пользовательскими историями, которые помогут на этапе кодирования проекта.
В этой статье мы можем начать пачкать руки и погрузиться в код. В этой статье рассматривается внутренняя часть приложения с использованием Python и PostgreSQL.
На протяжении всего этапа кодирования я буду постоянно ссылаться на диаграмму отношений классов:
И диаграмма классов:
И код в соответствии с ними. Поскольку между этими классами существует четкая иерархия, мы должны начинать сверху вниз, начиная с класса System, затем класса User и далее вниз до класса . Классы комментариев и Уведомления. В конце статьи проект будет иметь следующую структуру:
Проверьте мой репозиторий GitHub для получения полного кода. Репозиторий также содержит несколько модульных тестов для каждого класса с использованием pytest.
Подключение к PostgreSQL с помощью системы
Класс System будет классом общего назначения, и его обязанность будет в основном заключаться в обмене информацией между базой данных и другими классами. Важно, чтобы один класс нес эту ответственность, а не позволял каждому классу подключаться к базе данных самостоятельно, что может вызвать проблемы в будущем.
Для начала я создал базу данных под названием «управление задачами» в PostgreSQL. Если вы хотите продолжить, создайте аналогичный. Самый простой способ сделать это — щелкнуть правой кнопкой мыши вкладку «Базы данных» в Postgres.
Далее мы должны создать таблицы. У каждого класса будет своя таблица. Раньше я по следующим запросам создавал 5 таблиц и вставлял тестовую запись в каждую из них.
-- Table #1: public.users DROP TABLE IF EXISTS public.users; CREATE TABLE IF NOT EXISTS public.users ( user_id integer NOT NULL, name character varying(20) COLLATE pg_catalog."default", email character varying(20) COLLATE pg_catalog."default", password character varying(20) COLLATE pg_catalog."default", create_date timestamp without time zone, update_date timestamp without time zone, CONSTRAINT users_pkey PRIMARY KEY (user_id) ) TABLESPACE pg_default; ALTER TABLE IF EXISTS public.users OWNER to postgres; -- Table #2: public.projects DROP TABLE IF EXISTS public.projects; CREATE TABLE IF NOT EXISTS public.projects ( project_id integer NOT NULL, user_id integer NOT NULL, title character varying(30) COLLATE pg_catalog."default", description character varying(100) COLLATE pg_catalog."default", create_date timestamp without time zone, update_date timestamp without time zone, end_date timestamp without time zone, CONSTRAINT projects_pkey PRIMARY KEY (project_id), CONSTRAINT projects_user_id_fkey FOREIGN KEY (user_id) REFERENCES public.users (user_id) MATCH SIMPLE ON UPDATE NO ACTION ON DELETE CASCADE ) TABLESPACE pg_default; ALTER TABLE IF EXISTS public.projects OWNER to postgres; -- Table #3: public.tasks DROP TABLE IF EXISTS public.tasks; CREATE TABLE IF NOT EXISTS public.tasks ( task_id integer NOT NULL, project_id integer NOT NULL, title character varying(20) COLLATE pg_catalog."default" NOT NULL, description character varying(100) COLLATE pg_catalog."default" NOT NULL, status character varying(10) COLLATE pg_catalog."default", due_date timestamp without time zone, start_date timestamp without time zone, last_update_date timestamp without time zone, CONSTRAINT tasks_pkey PRIMARY KEY (task_id), CONSTRAINT tasks_project_id_fkey FOREIGN KEY (project_id) REFERENCES public.projects (project_id) MATCH SIMPLE ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT tasks_status_check CHECK (status::text = ANY (ARRAY['active'::character varying, 'inactive'::character varying, 'pending'::character varying]::text[])) ) TABLESPACE pg_default; ALTER TABLE IF EXISTS public.tasks OWNER to postgres; -- Table #4: public.comments DROP TABLE IF EXISTS public.comments; CREATE TABLE IF NOT EXISTS public.comments ( comment_id integer NOT NULL, task_id integer NOT NULL, content character varying(250) COLLATE pg_catalog."default", created_date timestamp without time zone, last_update_date timestamp without time zone, CONSTRAINT comments_pkey PRIMARY KEY (comment_id), CONSTRAINT comments_task_id_fkey FOREIGN KEY (task_id) REFERENCES public.tasks (task_id) MATCH SIMPLE ON UPDATE NO ACTION ON DELETE CASCADE ) TABLESPACE pg_default; ALTER TABLE IF EXISTS public.comments OWNER to postgres; -- Table #5: public.notifications DROP TABLE IF EXISTS public.notifications; CREATE TABLE IF NOT EXISTS public.notifications ( notification_id integer NOT NULL, task_id integer NOT NULL, content character varying(250) COLLATE pg_catalog."default", created_date timestamp without time zone, notify_date timestamp without time zone, last_update_date timestamp without time zone, CONSTRAINT notifications_pkey PRIMARY KEY (notification_id), CONSTRAINT notifications_task_id_fkey FOREIGN KEY (task_id) REFERENCES public.tasks (task_id) MATCH SIMPLE ON UPDATE NO ACTION ON DELETE CASCADE ) TABLESPACE pg_default; ALTER TABLE IF EXISTS public.notifications OWNER to postgres; insert into users(user_id, name, email, password, create_date, update_date) values (1, 'test', 'test', 'test', '2023-01-01', '2023-01-01'); INSERT INTO public.projects( project_id, user_id, title, description, create_date, update_date, end_date) VALUES (1, 1, 'test', 'test', '2023-01-01','2023-01-01', '2023-01-01'); INSERT INTO public.tasks( task_id, project_id, title, description, status, due_date, start_date, last_update_date) VALUES (1, 1, 'test', 'test', 'active', '2023-01-01', '2023-01-01', '2023-01-01'); INSERT INTO public.comments( comment_id, task_id, content, created_date, last_update_date) VALUES (1, 1, 'test', '2023-01-01', '2023-01-01'); INSERT INTO public.notifications( notification_id, task_id, content, created_date, notify_date, last_update_date) VALUES (1, 1, 'test', '2023-01-01', '2024-01-01', '2023-01-01');
Далее создадим конструктор для класса System, он будет выглядеть так:
from psycopg2 import connect import sys import os import datetime sys.path.append('/projects/task_management') class System: def __init__(self, host = '***.***.*.***', user = 'postgres', password = ****, port = 5432, database = 'task_management') -> None: ''' Constructor for System class Args: host: host ip address(default: my local ip address) user: username (default: postgres) password: password (default: ****) port: port number (default: 5432) database: database name (default: task_management) Returns: None ''' self.host = host self.user = user self.password = password self.port = port self.database = database
Далее мы можем создать метод для помощи в подключении к Postgres:
from psycopg2 import connect import sys import os import datetime sys.path.append('/projects/task_management') class System: def __init__(self, host = '***.***.*.***', user = 'postgres', password = ****, port = 5432, database = 'task_management') -> None: ''' Constructor for System class Args: host: host ip address(default: my local ip address) user: username (default: postgres) password: password (default: ****) port: port number (default: 5432) database: database name (default: task_management) Returns: None ''' self.host = host self.user = user self.password = password self.port = port self.database = database def connect_to_db(self): ''' Connect to PostgreSQL database Args: None Returns: conn: connection to database ''' try: conn = connect(host = self.host, user = self.user, password = self.password, port = self.port, database = self.database) return conn except: print("Error connecting to database")
Давайте проверим, что мы можем подключиться к Postgres:
import sys sys.path.append('/projects/task_management') from classes.system import System # Create a system instance system = System() # Connect to the database conn = system.connect_to_db() # Verify connection if conn.status == 1: print('Connected to database successfully') else: print('Failed to connect to database')
Выход:
Connected to database successfully
Затем я добавил еще несколько методов в класс System:
- grab_max_object_id(object) — в зависимости от объекта этот метод будет подключаться к базе данных и получать последний идентификатор объекта. Это будет чрезвычайно полезный метод.
- insert_user() , delete_user() и update_user_attributes() — эти методы будут подключаться к базе данных и выполнять запросы на вставку, удаление и обновление таблица пользователей.
Текущая система будет выглядеть так:
from psycopg2 import connect import sys import os import datetime sys.path.append('/projects/task_management') class System: def __init__(self, host = '***.***.*.***', user = 'postgres', password = ****, port = 5432, database = 'task_management') -> None: ''' Constructor for System class Args: host: host ip address(default: my local ip address) user: username (default: postgres) password: password (default: 1365) port: port number (default: 5432) database: database name (default: task_management) Returns: None ''' self.host = host self.user = user self.password = password self.port = port self.database = database def connect_to_db(self): ''' Connect to PostgreSQL database Args: None Returns: conn: connection to database ''' try: conn = connect(host = self.host, user = self.user, password = self.password, port = self.port, database = self.database) return conn except: print("Error connecting to database") def grab_max_object_id(self, object): ''' Grab the max object id from the database Args: None Returns: max_id: max object id ''' conn = self.connect_to_db() cursor = conn.cursor() if object == 'user': cursor.execute("SELECT MAX(user_id) FROM users") if object == 'project': cursor.execute("SELECT MAX(project_id) FROM projects") if object == 'task': cursor.execute("SELECT MAX(task_id) FROM tasks") if object == 'comment': cursor.execute("SELECT MAX(comment_id) FROM comments") if object == 'notification': cursor.execute("SELECT MAX(notification_id) FROM notifications") max_id = cursor.fetchone()[0] return max_id ## User Related Methods ## def insert_user(self, id, name, email, password, create_date, update_date): ''' Insert user to database to users table Args: id: user id name: user name email: user email password: user password create_date: user create date update_date: user update date Returns: None ''' conn = self.connect_to_db() cur = conn.cursor() cur.execute("INSERT INTO users (user_id, name, email, password, create_date, update_date) VALUES (%s, %s, %s, %s, %s, %s)", (id, name, email, password, create_date, update_date)) conn.commit() cur.close() conn.close() def delete_user(self, user_id): ''' Delete user from database Args: user_id: user id Returns: None ''' conn = self.connect_to_db() cur = conn.cursor() cur.execute("DELETE FROM users WHERE user_id = %s", (user_id,)) conn.commit() cur.close() conn.close() def update_user_attributes(self, user_id, name, email, password): ''' Update user attributes Args: user_id: user id name: user name email: user email password: user password Returns: None ''' conn = self.connect_to_db() cursor = conn.cursor() query = "update users set name = %s, email = %s,password = %s, update_date = %s where user_id = %s" cursor.execute(query,(name, email, password, datetime.datetime.today(), user_id)) conn.commit()
На данный момент этого достаточно, мы вернемся и будем постоянно улучшать этот класс. А пока давайте перейдем к классу User.
Классы пользователей и проектов
Далее я создам первый набросок для класса User в соответствии с атрибутами и методами диаграммы классов.
import datetime import sys sys.path.append('/projects/task_management') from classes.project import Project from classes.system import System class User: # insert user to database system = System() # User constructor def __init__(self, name: str, email: str, password: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today()) -> None: ''' Constructor for User class Args: name: user name email: user email password: user password create_date: user create date (default: today) update_date: user update date (default: today) Returns: None Notes: This constructor will insert the user to the database ''' self._name = name self._email = email self._password = password self.create_date = create_date self.update_date = update_date # Grab the last id from the database and add 1 to it id = User.system.grab_max_object_id(object='user') + 1 # Insert user to db User.system.insert_user(id, name, email, password, create_date, update_date) self._id = id # Create an empty list for projects self.projects = []
Здесь есть несколько замечаний.
Во-первых, обратите внимание, что класс System является переменной класса внутри класса User. Это свяжет их обоих и позволит классу User взаимодействовать с базой данных через методы класса System.
Во-вторых, вместо жесткого кодирования нового идентификатора пользователя каждый раз, когда мы создаем пользователя, я использовал класс System, чтобы получить максимальный идентификатор пользователя из таблицы пользователей и добавить к нему 1. Таким образом, мы можем создать столько пользователей, сколько захотим, не нарушая первичный ключ в базе данных.
В-третьих, обратите внимание, что каждый раз, когда мы будем создавать пользователя, конструктор __init__ будет запускать system.insert_user(…).
Другими словами, когда мы создаем пользователя с помощью Python, этот же пользователь будет автоматически вставлен в таблицу пользователей в Postgres.
В-четвертых, обратите внимание, что одним из атрибутов пользователя является список проектов. Это позволит нам позже соединить классы User и Project, поскольку каждый отдельный пользователь может иметь несколько проектов.
Получить и установить методы для пользователя
Далее добавим несколько методов get и set. Обратите внимание, что методы set также будут использовать класс System, чтобы не только изменять атрибуты отдельного пользователя в Python, но и обновлять их в Postgres.
import datetime import sys sys.path.append('/projects/task_management') from classes.project import Project from classes.system import System class User: # insert user to database system = System() # User constructor def __init__(self, name: str, email: str, password: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today()) -> None: ''' Constructor for User class Args: name: user name email: user email password: user password create_date: user create date (default: today) update_date: user update date (default: today) Returns: None Notes: This constructor will insert the user to the database ''' self._name = name self._email = email self._password = password self.create_date = create_date self.update_date = update_date # Grab the last id from the database and add 1 to it id = User.system.grab_max_object_id(object='user') + 1 # Insert user to db User.system.insert_user(id, name, email, password, create_date, update_date) self._id = id # Create an empty list for projects self.projects = [] # Get methods for id, name, email and password @property def id(self): return self._id @property def name(self): return self._name @property def email(self): return self._email @property def password(self): return self._password # Set methods for name, email and password # Each set method updates the database as well @name.setter def name(self, new_name): if isinstance(new_name, str): User.system.update_user_attributes(self.id, new_name, self.email, self.password) self._name = new_name else: raise TypeError("Name must be a text!") @email.setter def email(self, new_email): if isinstance(new_email, str): User.system.update_user_attributes(self.id, self.name, new_email, self.password) self._email = new_email else: raise TypeError("Email must be a text!") @password.setter def password(self, new_password): if isinstance(new_password, str): User.system.update_user_attributes(self.id, self.name, self.email, new_password) self._password = new_password else: raise TypeError("Password must be a text!")
Добавление create_project() и delete_project() для пользователя
И, наконец, согласно диаграмме классов класс User отвечает за управление проектами. Добавлю еще два метода создания проекта и удаления проекта. Класс User теперь будет выглядеть следующим образом:
import datetime import sys sys.path.append('/projects/task_management') from classes.project import Project from classes.system import System class User: # insert user to database system = System() # User constructor def __init__(self, name: str, email: str, password: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today()) -> None: ''' Constructor for User class Args: name: user name email: user email password: user password create_date: user create date (default: today) update_date: user update date (default: today) Returns: None Notes: This constructor will insert the user to the database ''' self._name = name self._email = email self._password = password self.create_date = create_date self.update_date = update_date # Grab the last id from the database and add 1 to it id = User.system.grab_max_object_id(object='user') + 1 # Insert user to db User.system.insert_user(id, name, email, password, create_date, update_date) self._id = id # Create an empty list for projects self.projects = [] # Get methods for id, name, email and password @property def id(self): return self._id @property def name(self): return self._name @property def email(self): return self._email @property def password(self): return self._password # Set methods for name, email and password # Each set method updates the database as well @name.setter def name(self, new_name): if isinstance(new_name, str): User.system.update_user_attributes(self.id, new_name, self.email, self.password) self._name = new_name else: raise TypeError("Name must be a text!") @email.setter def email(self, new_email): if isinstance(new_email, str): User.system.update_user_attributes(self.id, self.name, new_email, self.password) self._email = new_email else: raise TypeError("Email must be a text!") @password.setter def password(self, new_password): if isinstance(new_password, str): User.system.update_user_attributes(self.id, self.name, self.email, new_password) self._password = new_password else: raise TypeError("Password must be a text!") # Create Project def create_project(self, title: str, description: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today(), end_date = datetime.datetime.today() + datetime.timedelta(days=365)): ''' Create a new project for the user Args: title: project title description: project description create_date: project create date (default: today) update_date: project update date (default: today) end_date: project end date (default: today + 365 days) Returns: project_id Notes: - This method will insert the project to the database - This method will append the project to the User.projects list ''' self.projects.append(Project(user_id = self.id, title=title, description=description, create_date=create_date, update_date=update_date, end_date=end_date)) return self.projects[-1] # delete project def delete_project(self, project_title): ''' Delete a project from the user Args: project_title: project title Returns: None Notes: - This method will delete the project from the database - This method will delete the project from the User.projects list ''' for project in self.projects: if project.title == project_title: self.projects.remove(project) User.system.delete_project(project.id) break
Важные моменты, на которые следует обратить внимание:
Во-первых, метод create_project() вставит экземпляр класса Project в список self.projects. Опять же, это свяжет отдельного пользователя с таким количеством экземпляров класса Project, сколько необходимо.
Во-вторых, вскоре вы заметите, что, подобно классу User, класс Project также будет автоматически вставлять Project в таблицу проектов в Postgres всякий раз, когда создается экземпляр.
Другими словами, после создания класса Project отдельный пользователь сможет создать экземпляр класса Project, и этот экземпляр будет записан в базу данных с правильные ограничения первичного и внешнего ключа.
Класс проекта
Прежде чем показать демонстрацию, я быстро покажу начальную реализацию класса Project:
import datetime import sys sys.path.append('/projects/task_management') from classes.system import System class Project: # Create a system instance system = System() def __init__(self, user_id, title: str, description: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today(), end_date = datetime.datetime.today() + datetime.timedelta(days=365)) -> None: ''' Constructor for Project class Args: title: project title description: project description create_date: project create date (default: today) update_date: project update date (default: today) end_date: project end date (default: today + 365 days) Returns: None Notes: - This constructor will insert the project to the database. - Only User class can create a project. ''' self._title = title self._description = description self._end_date = end_date self.create_date = create_date self.update_date = update_date self.user_id = user_id # Grab the last id from the database and add 1 to it last_id = Project.system.grab_max_object_id(object='project') id = last_id + 1 # Insert project to db Project.system.insert_project(id, user_id, title, description, create_date, update_date, end_date) self._id = id # Create an empty list for projects self.tasks = [] # Get methods for name, email and password @property def id(self): return self._id @property def title(self): return self._title @property def description(self): return self._description @property def end_date(self): return self._end_date # Set methods for name, email and password # Each set method updates the database as well @title.setter def title(self, new_title): if isinstance(new_title, str): Project.system.update_project_attributes(self.id, new_title, self.description, self.end_date) self._title = new_title else: raise TypeError("Title must be a text!") @description.setter def description(self, new_description): if isinstance(new_description, str): Project.system.update_project_attributes(self.id, self.title, new_description, self.end_date) self._description = new_description else: raise TypeError("Description must be a text!") @end_date.setter def end_date(self, new_end_date): if isinstance(new_end_date, datetime.datetime): Project.system.update_project_attributes(self.id, self.title, self.description, new_end_date) self._end_date = new_end_date else: raise TypeError("End date must be a date!")
В этом классе нет ничего нового, дизайн очень похож на класс User.
Использование класса пользователя и проекта
Давайте быстро продемонстрируем, как Пользователь, Проект и Postgres взаимодействуют друг с другом.
import sys sys.path.append('/projects/task_management') from classes.system import System from classes.user import User from classes.project import Project # Create a system instance system = System() # Create a user john = User(name = 'John',email='[email protected]', password='123456') # Using the new user to create a project johns_project = john.create_project(title='Project 1', description='This is a project')
Как мы видим, мы можем использовать только что созданного пользователя для создания проекта, давайте проверим, вставлены ли в базу данных и новый пользователь, и новый проект. Давайте возьмем идентификатор пользователя Джона:
print(john.id) # 2
Теперь поищем его в базе данных. Сначала таблица пользователей:
Вторая таблица проектов:
Большой. Похоже, все работает, и Пользователь, Проект и Postgres тесно связаны друг с другом.
Добавление create_task() и delete_task() в проект
Далее, аналогично классу User, я добавлю в класс Project методы для создания задачи удаления. Полный класс Project выглядит следующим образом:
import datetime import sys sys.path.append('/projects/task_management') from classes.system import System from classes.task import Task class Project: # Create a system instance system = System() def __init__(self, user_id, title: str, description: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today(), end_date = datetime.datetime.today() + datetime.timedelta(days=365)) -> None: ''' Constructor for Project class Args: title: project title description: project description create_date: project create date (default: today) update_date: project update date (default: today) end_date: project end date (default: today + 365 days) Returns: None Notes: - This constructor will insert the project to the database. - Only User class can create a project. ''' self._title = title self._description = description self._end_date = end_date self.create_date = create_date self.update_date = update_date self.user_id = user_id # Grab the last id from the database and add 1 to it last_id = Project.system.grab_max_object_id(object='project') id = last_id + 1 # Insert project to db Project.system.insert_project(id, user_id, title, description, create_date, update_date, end_date) self._id = id # Create an empty list for projects self.tasks = [] # Get methods for name, email and password @property def id(self): return self._id @property def title(self): return self._title @property def description(self): return self._description @property def end_date(self): return self._end_date # Set methods for name, email and password # Each set method updates the database as well @title.setter def title(self, new_title): if isinstance(new_title, str): Project.system.update_project_attributes(self.id, new_title, self.description, self.end_date) self._title = new_title else: raise TypeError("Title must be a text!") @description.setter def description(self, new_description): if isinstance(new_description, str): Project.system.update_project_attributes(self.id, self.title, new_description, self.end_date) self._description = new_description else: raise TypeError("Description must be a text!") @end_date.setter def end_date(self, new_end_date): if isinstance(new_end_date, datetime.datetime): Project.system.update_project_attributes(self.id, self.title, self.description, new_end_date) self._end_date = new_end_date else: raise TypeError("End date must be a date!") # create task def create_task(self, title: str, description: str, create_date = datetime.datetime.today(), update_date = datetime.datetime.today(), end_date = datetime.datetime.today() + datetime.timedelta(days=365)): ''' Create a task for the project Args: title: task title description: task description create_date: task create date (default: today) update_date: task update date (default: today) end_date: task end date (default: today + 365 days) Returns: Task object Notes: - This method will insert the task to the database. - Only Project class can create a task. ''' self.tasks.append(Task(self.id, title, description, create_date, update_date, end_date)) return self.tasks[-1] # delete task def delete_task(self, task_id): for task in self.tasks: if task.task_id == task_id: self.tasks.remove(task) Project.system.delete_task(task.task_id) break
Опять же, логика очень похожа на класс User, в настоящее время экземпляр класса Project может создать новый экземпляр класса Task. . Этот экземпляр будет автоматически вставлен в таблицу задач в базе данных.
Для краткости этой статьи я остановлюсь здесь, но взгляните на полный код в моем репозитории GitHub и обратите внимание, что структура классов остается примерно одинаковой, все они тесно связаны между собой, класс System и базу данных Postgres таким образом, чтобы свести к минимуму ответственность пользователя и обеспечить максимальную согласованность базы данных.