当前位置： 首页 > news >正文

python: DDD using postgeSQL and SQL Server

news 来源：原创 2025/5/23 1:09:12

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, relationship
 
 
Base = 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 Teacher
 
class 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_size
            query = 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 []  # 返回空列表而不是 None
        finally:
            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 None
        finally:
            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()
            raise
        finally:
            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_name
                existing_teacher.last_name = teacher.last_name
                existing_teacher.gender = teacher.gender
                existing_teacher.tel_no = teacher.tel_no
                existing_teacher.school_id = teacher.school_id
                session.commit()
        except:
            session.rollback()
            raise
        finally:
            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()
            raise
        finally:
            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 count
        except Exception as e:
            print(f"Error getting total count: {e}")
            return 0
        finally:
            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 MssqlHelper
 
Base = 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 SQLAlchemyError
 
class 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_name
            teacher_model.TeacherLastName = teacher.last_name
            teacher_model.TeacherGender = teacher.gender
            teacher_model.TeacherTelNo = teacher.tel_no
            teacher_model.TeacherSchoolId = teacher.school_id
            session.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_name
                teacher.TeacherLastName = last_name
                teacher.TeacherGender = gender
                teacher.TeacherTelNo = tel_no
                teacher.TeacherSchoolId = school_id
 
            # 更新关联学校信息
            school = session.query(School).filter_by(SchoolId=school_id).first()
            if school:
                school.SchoolName = school_name
                school.SchoolTelNo = school_tel_no
 
            session.commit()
        except Exception as e:
            session.rollback()
            raise e
        finally:
            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_size
        query = 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

输出：

相关文章：

  • 【STM32】STM32系列产品以及新手入门的STM32F103
  • 深度学习PyTorch之13种模型精度评估公式及调用方法
  • 头歌作业-数据库实验一：数据库和数据表的建立，修改和删除
  • 大模型——使用 Embedding 模型和向量数据库的 Spring AI RAG
  • Linux下磁盘读写流
  • 《几何原本》命题I.12
  • API和SDK
  • 护照阅读器在汽车客运站流程中的应用
  • Excel表格打印 第二页边框隔断
  • 在Spring Boot + MyBatis中优雅处理多表数据清洗：基于XML的配置化方案
  • Android Coil3缩略图、默认占位图placeholder、error加载错误显示，Kotlin（5）
  • #如何改变怂怂懦弱的气质（2）
  • 【Linux内核系列】：进入文件系统的世界
  • µCOS-III从入门到精通 第七章（任务调度）
  • 算法日记34：14届蓝桥C++B接龙数列(动态规划DP)
  • 安全见闻之网络安全新兴术语
  • ThreadLocal
  • 马尔科夫不等式和切比雪夫不等式
  • 为AI聊天工具添加一个知识系统 之138 设计重审 2 文章学 之2
  • Linux基础 IO 和文件
  • 购物网站建设要多少钱/电商网站建设 网站定制开发
  • 免费网站建设协议/友情链接平台网站
  • 网站开发策划个人简历/成都seo网络优化公司
  • 南山品牌网站建设企业/爱站网长尾关键词挖掘工具
  • 做网站注册页面/软件外包网站
  • 大连开发区社保网站/免费发广告的网站大全