python: DDD using postgeSQL and SQL Server

postgreSQL

注意：

# psycopg 2 驱动的连接字符串
#engine = create_engine('postgresql://post:geovindu@localhost:5433/TechnologyGame')
#Session = sessionmaker(bind=engine)# 使用 psycopg3 驱动的连接字符串 
#engine = create_engine('postgresql+psycopg://user:geovindu@localhost:5432/TechnologyGame')
#Session = sessionmaker(bind=engine)

create table School  -- 創建表
(SchoolId char(5) NOT NULL PRIMARY KEY,   SchoolName varchar(500) NOT NULL DEFAULT '', SchoolTelNo  varchar(8)  NULL DEFAULT '' 
);create table Teacher  -- 創建表
(TeacherId char(5) NOT NULL ,TeacherFirstName varchar(100) NOT NULL DEFAULT '', TeacherLastName varchar(20) NOT NULL DEFAULT '',TeacherGender char(2) NOT NULL DEFAULT '',TeacherTelNo  varchar(8)  NULL DEFAULT '',TeacherSchoolId  char(5) NOT NULL DEFAULT '', PRIMARY KEY (TeacherId),   -- 主鍵FOREIGN KEY(TeacherSchoolId) REFERENCES School(SchoolId)  -- 外鍵
);  

项目结构：

# encoding: utf-8
# 版权所有 2025 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : PyCharm 2023.1 python 3.11
# OS        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  oracle 21c Neo4j
# Datetime  : 2025/3/6 22:51
# User      : geovindu
# Product   : PyCharm
# Project   : pypostgreSQLDDDOrmDemo
# File      : teacher.py
# explain   : 学习
from sqlalchemy import create_engine, Column, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, declarative_base, relationshipBase = declarative_base()''' '''
class SchoolModel(Base):"""必须小写字母"""__tablename__ = 'school'schoolid = Column(String(5), primary_key=True)schoolname = Column(String(500), default='')schooltelno = Column(String(8), default='')teachers = relationship("TeacherModel", back_populates="school")class TeacherModel(Base):"""必须小写字母"""__tablename__ = 'teacher'teacherid = Column(String(5), primary_key=True)teacherfirstname = Column(String(100), default='')teacherlastname = Column(String(20), default='')teachergender = Column(String(2), default='')teachertelno = Column(String(8), default='')teacherschoolid = Column(String(5), ForeignKey(SchoolModel.schoolid))school = relationship("SchoolModel", back_populates="teachers")

# encoding: utf-8
# 版权所有 2025 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : PyCharm 2023.1 python 3.11
# OS        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  oracle 21c Neo4j
# Datetime  : 2025/3/6 22:58
# User      : geovindu
# Product   : PyCharm
# Project   : pypostgreSQLDDDOrmDemo
# File      : teacher.py
# explain   : 学习
from domain.repositories.teacher import TeacherRepository
from infrastructure.database.postgresqlHelper import PostgresqlHelper
from infrastructure.model.teacher import TeacherModel
from domain.entities.teacher import Teacherclass TeacherRepositoryImpl(TeacherRepository):""""""def __init__(self):self.sesion = PostgresqlHelper()def get_all(self, page=1, page_size=10, keyword=''):""":param page::param page_size::param keyword::return:"""session = self.sesion.getSession()try:offset = (page - 1) * page_sizequery = session.query(TeacherModel)print(query)#query = query.order_by(TeacherModel.teacherid)#total = query.count()#print("total", total)if keyword:query = query.filter((TeacherModel.teacherfirstname.contains(keyword)) |(TeacherModel.teacherlastname.contains(keyword)))teachers =query.offset(offset).limit(page_size).all() # query.offset(offset).limit(page_size).all()print("data teachers",teachers)return [Teacher(teacher.teacherid, teacher.teacherfirstname, teacher.teacherlastname, teacher.teachergender, teacher.teachertelno, teacher.teacherschoolid) for teacher in teachers]except Exception as ex:print(ex.__str__())print(f"Error fetching teachers: {ex}")#return []  # 返回空列表而不是 Nonefinally:session.close()def get_by_id(self, teacher_id):""":param teacher_id::return:"""session = self.sesion.getSession()try:teacher = session.query(TeacherModel).filter_by(teacher_id=teacher_id).first()if teacher:return Teacher(teacher.teacherid, teacher.teacherfirstname, teacher.teacherlastname, teacher.teachergender, teacher.teachertelno, teacher.teacherschoolid)return Nonefinally:session.close()def add(self, teacher: Teacher):""":param teacher::return:"""session = self.sesion.getSession()try:new_teacher = TeacherModel(teacher_id=teacher.teacher_id, first_name=teacher.first_name, last_name=teacher.last_name, gender=teacher.gender, tel_no=teacher.tel_no, school_id=teacher.school_id)session.add(new_teacher)session.commit()except:session.rollback()raisefinally:session.close()def update(self, teacher: Teacher):""":param teacher::return:"""session = self.sesion.getSession()try:existing_teacher = session.query(TeacherModel).filter_by(teacher_id=teacher.teacher_id).first()if existing_teacher:existing_teacher.first_name = teacher.first_nameexisting_teacher.last_name = teacher.last_nameexisting_teacher.gender = teacher.genderexisting_teacher.tel_no = teacher.tel_noexisting_teacher.school_id = teacher.school_idsession.commit()except:session.rollback()raisefinally:session.close()def delete(self, teacher_id):""":param teacher_id::return:"""session = self.sesion.getSession()try:teacher = session.query(TeacherModel).filter_by(teacher_id=teacher_id).first()if teacher:session.delete(teacher)session.commit()except:session.rollback()raisefinally:session.close()def get_total_count(self, keyword=''):""":param keyword::return:"""session = self.sesion.getSession()try:query = session.query(TeacherModel)if keyword:query = query.filter((TeacherModel.first_name.contains(keyword)) |(TeacherModel.last_name.contains(keyword)))count = query.count()return countexcept Exception as e:print(f"Error getting total count: {e}")return 0finally:session.close()

sql server:

create table School  -- 創建表
(SchoolId char(5) NOT NULL PRIMARY KEY,  -- 學校編號SchoolName nvarchar(500) NOT NULL  DEFAULT '',  --  學校名稱',SchoolTelNo  varchar(8)  NULL DEFAULT '' , -- 電話號碼   ); 
create table Teacher  -- 創建表
(TeacherId char(5) NOT NULL , --'主鍵primary key，學生編號',TeacherFirstName nvarchar(100) NOT NULL DEFAULT '',  -- ' 名',TeacherLastName nvarchar(20) NOT NULL DEFAULT '', -- ' 姓',TeacherGender char(2) NOT NULL DEFAULT '', -- '性別',TeacherTelNo  varchar(8)  NULL DEFAULT '',  --'電話號碼',TeacherSchoolId  char(5) NOT NULL DEFAULT '',  -- '外鍵 foreign key 學校ID',     PRIMARY KEY (TeacherId),   -- 主鍵FOREIGN KEY(TeacherSchoolId) REFERENCES School(SchoolId)  -- 外鍵
)

项目结构：

# encoding: utf-8
# 版权所有 2025 ©涂聚文有限公司 ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : PyCharm 2023.1 python 3.11
# OS        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  oracle 21c Neo4j
# Datetime  : 2025/2/20 20:46
# User      : geovindu
# Product   : PyCharm
# Project   : pyMsSqlDDDOrmDemo
# File      : infrastructure/model/teacher.py
# explain   : 学习
from sqlalchemy import create_engine, Column, String, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
from ..database.mssqlHelper import MssqlHelperBase = declarative_base()class SchoolModel(Base):"""基础设施层（Infrastructure）数据库交互"""__tablename__ = 'School'SchoolId = Column(String(5), primary_key=True)SchoolName = Column(String(500), default='')SchoolTelNo = Column(String(8), default='')teachers = relationship("TeacherModel", back_populates="school")class TeacherModel(Base):"""基础设施层（Infrastructure）数据库交互"""__tablename__ = 'Teacher'TeacherId = Column(String(5), primary_key=True)TeacherFirstName = Column(String(100), default='')TeacherLastName = Column(String(20), default='')TeacherGender = Column(String(2), default='')TeacherTelNo = Column(String(8), default='')TeacherSchoolId = Column(String(5), ForeignKey('School.SchoolId'))school = relationship("SchoolModel", back_populates="teachers")

# encoding: utf-8
# 版权所有 2025 ©涂聚文有限公司 ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author    : geovindu,Geovin Du 涂聚文.
# IDE       : PyCharm 2023.1 python 3.11
# OS        : windows 10
# database  : mysql 9.0 sql server 2019, postgreSQL 17.0  oracle 21c Neo4j
# Datetime  : 2025/2/20 20:47
# User      : geovindu
# Product   : PyCharm
# Project   : pyMsSqlDDDOrmDemo
# File      : infrastructure/repositories/teacher.py
# explain   : 学习from sqlalchemy import create_engine, Column, String, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
from domain.entities.teacher import Teacher
from domain.entities.school import School
from ..database.mssqlHelper import MssqlHelper
from ..model.teacher import TeacherModel
from sqlalchemy.exc import SQLAlchemyErrorclass TeacherRepository:"""基础设施层（Infrastructure）实现存储库接口"""def __init__(self):""""""self._Session = MssqlHelper()def add(self, teacher: Teacher):""":param teacher::return:"""session = self._Session.getSession()teacher_model = TeacherModel(TeacherId=teacher.teacher_id, TeacherFirstName=teacher.first_name,TeacherLastName=teacher.last_name, TeacherGender=teacher.gender,TeacherTelNo=teacher.tel_no, TeacherSchoolId=teacher.school_id)session.add(teacher_model)session.commit()session.close()def update(self, teacher: Teacher):""":param teacher::return:"""session = self._Session.getSession()teacher_model = session.query(TeacherModel).filter_by(TeacherId=teacher.teacher_id).first()if teacher_model:teacher_model.TeacherFirstName = teacher.first_nameteacher_model.TeacherLastName = teacher.last_nameteacher_model.TeacherGender = teacher.genderteacher_model.TeacherTelNo = teacher.tel_noteacher_model.TeacherSchoolId = teacher.school_idsession.commit()session.close()def update_teacher_and_school(self, teacher_id, first_name, last_name, gender, tel_no, school_id, school_name,school_tel_no):""":param teacher_id::param first_name::param last_name::param gender::param tel_no::param school_id::param school_name::param school_tel_no::return:"""session = self._Session.getSession()try:# 更新老师信息teacher = session.query(Teacher).filter_by(TeacherId=teacher_id).first()if teacher:teacher.TeacherFirstName = first_nameteacher.TeacherLastName = last_nameteacher.TeacherGender = genderteacher.TeacherTelNo = tel_noteacher.TeacherSchoolId = school_id# 更新关联学校信息school = session.query(School).filter_by(SchoolId=school_id).first()if school:school.SchoolName = school_nameschool.SchoolTelNo = school_tel_nosession.commit()except Exception as e:session.rollback()raise efinally:session.close()def delete(self, teacher_id: str):""":param teacher_id::return:"""session = self._Session.getSession()teacher_model = session.query(TeacherModel).filter_by(TeacherId=teacher_id).first()if teacher_model:session.delete(teacher_model)session.commit()session.close()def get_all(self, page: int, page_size: int, search_query: str = ""):""":param page::param page_size::param search_query::return:"""session = self._Session.getSession()query = session.query(TeacherModel)if search_query:query = query.filter((TeacherModel.TeacherId.contains(search_query)) |(TeacherModel.TeacherFirstName.contains(search_query)) |(TeacherModel.TeacherLastName.contains(search_query)) |(TeacherModel.TeacherGender.contains(search_query)) |(TeacherModel.TeacherTelNo.contains(search_query)) |(TeacherModel.TeacherSchoolId.contains(search_query)))offset = (page - 1) * page_sizequery = query.order_by(TeacherModel.TeacherId)  # 这里以 TeacherId 为例进行排序，你可以根据实际需求修改排序字段total = query.count()#print(" teacher total", total)teacher_models = query.offset(offset).limit(page_size).all()session.close()return [Teacher(teacher.TeacherId, teacher.TeacherFirstName, teacher.TeacherLastName,teacher.TeacherGender, teacher.TeacherTelNo, teacher.TeacherSchoolId)for teacher in teacher_models]def get_total_count(self, search_query: str = ""):""":param search_query::return:"""session = self._Session.getSession()query = session.query(TeacherModel)if search_query:query = query.filter((TeacherModel.TeacherId.contains(search_query)) |(TeacherModel.TeacherFirstName.contains(search_query)) |(TeacherModel.TeacherLastName.contains(search_query)) |(TeacherModel.TeacherGender.contains(search_query)) |(TeacherModel.TeacherTelNo.contains(search_query)) |(TeacherModel.TeacherSchoolId.contains(search_query)))count = query.count()session.close()return count

输出：