Dividing Functionality with Classes

from typing import Callable, Tuple, List, Sequence
from pandas import DataFrame

from pandas import DataFrame  
from db import insert_df_pg, create_conn  
from utils import turn_tuple_into_df  
from psycopg2.extensions import connection  
  
  
INSERT_TB = 'insert_table' # 处理好数据存入的数据库表名  
INSERT_SC = 'insert_sc' # 处理好数据存入的schema名称  
INSERT_DB = 'insert_db' # 处理好数据存入的database名称  
INSERT_COL_TYEPS = {'id': 'int4', 'flag': 'float4', 'time':'timestamp'} # 插入数据库表的字段名与字段类型；因为会被下方多个函数引用，所有定义成全局变量  
  
  
def check_tb() -> bool:  
    '''  
    检查用于插入处理好的数据表是否存在以及是否符合要求  
    '''  
  
    if conn is None: # 如果没有生成数据库链接对象则生成  
        conn = create_conn(db_type='pg')  
    cur = conn.cursor()  
  
    # 生成查询数据表信息的sql语句  
    sql_info = f'''select column_name, udt_name from information_schema.columns  
    where table_schema = '{INSERT_SC}' and table_name = '{INSERT_TB}' '''  
      
    # 获取表字段，返回的是嵌套的tuple结构((), (), ())；如过为空则是一个空tuple  
    with conn, cur:  
        cur.execute(sql_info)  
        info_raw = cur.fetchall()  
    conn.close()  
  
    if not info_raw: # 返回的是空tuple说明不存在这个表  
        return False  
  
    info_df = turn_tuple_into_df(info_raw, col_names=['col_name', 'data_type']) # 转化为dataframe  
      
    # 检查是否有字段缺失  
    df_cols = info_df.columns.to_list() # 输入的数据列名  
    missed_cols = set(INSERT_COL_TYEPS.keys()).difference(df_cols) # 查看缺失哪些字段      
    if missed_cols:  
        return False  
  
    # 检查字段是否符合要求  
    col_type_check = info_df.apply(lambda sr: INSERT_COL_TYEPS[sr['col_name']] == sr['data_type'], axis=1)  
    if not col_type_check.all(axis=0):   
        return False  
  
    return True # 都没有问题，返回True  
  
# 剩余代码见文件`data_save.py`  
...  

# 完成代码见`data_save_with_class.py`  
class DataMove:  
    '''  
    用于旧数据迁移  
    '''  
    def __init__(self, db, sc, tb):  
        '''  
        :param db: 迁移表的database  
        :param sc: 迁移表的schema  
        :param tb: 迁移表的表名  
        '''  
        self.db = db  
        self.sc = sc  
        self.tb = tb  
                  
    def init_copy_tb(self):  
        '''  
        初始化用于储存迁移数据的表  
  
        '''  
        tb_init_ins = TbInit(self.db, self.sc, self.tb) # 实例化，为后续使用"准备数据表"功能做准备  
        is_tb_ready = tb_init_ins.check_tb() # 检查迁移表是否满足要求  
        if not is_tb_ready: # 不满足  
            tb_init_ins.drop_tb() # 删除迁移表  
            tb_init_ins.create_tb() # 按要求创建迁移表  
    ...  

class TbInit:  
    '''  
    该类用于准备数据储存的数据表  
    '''  
  
    # 类属性，为原始的数据储存表的信息  
    origin_tb = INSERT_TB  
    origin_sc = INSERT_SC  
    origin_db = INSERT_DB  
  
  
    def __init__(self, db: str = INSERT_DB, sc: str = INSERT_SC, tb: str = INSERT_TB):  
        '''  
        实例化时输入需要数据库表信息，不局限于全局变量中的数据库表，方便复用  
        '''  
        self.db = db  
        self.sc = sc  
        self.tb = tb  
  
    def check_tb(self, conn: connection = None) -> bool:  
        # 中间代码省略，具体见`data_save.py`  
        ...  
  
    @classmethod  
    def get_origin_tb_info(cls) -> Union[list[tuple] , tuple[tuple, ...]]:  
        '''  
        类方法，返回原始数据表的信息  
        '''  
        if conn is None: # 如果没有生成数据库链接对象则生成  
            conn = create_conn(db_type='pg')  
        cur = conn.cursor()  
  
        # 生成查询数据表信息的sql语句  
        sql_info = f'''select column_name, udt_name from information_schema.columns  
        where table_schema = '{cls.origin_sc}' and table_name = '{cls.origin_tb}' ''' # 可以看到，类方法可以调用类属性origin_tb  
  
        with conn, cur:  
            cur.execute(sql_info)  
            res = cur.fetchall()  
          
        return res  
  
    @staticmethod  
    def print_class_desc(): # 静态方法第一个参数不用填写cls/slef  
        '''  
        静态方法，打印这个类的功能  
        '''  
  
        print('TbInit是用于....')  

TbInit.print_class_desc() # 不用实例化直接调用静态方法  
TbInit.get_origin_tb_info() # 不用实例化直接调用类方法  
TbInit.origin_db # 不用实例化直接调用类属性  
tb_init_ins = TbInit('ins_db', 'ins_sc','ins_tb') # 实例化  
tb_init_ins.check_tb() # 实例化后调用实例方法  
tb_init_ins.db # 实例化后调用实例属性  

class TbInit:  
    '''  
    该类用于准备数据储存的数据表  
    '''  
  
    def __init__(self, db: str = INSERT_DB, sc: str = INSERT_SC, tb: str = INSERT_TB):  
        self.db = db # 绑定在self上的属性  
        ...  
  
    def check_tb(self, conn: connection = None) -> bool:  
        self.drop_tb() # 改写check_tb用来演示"绑定"和"共享"  
        self.db  
  
    def drop_tb(self, conn: connection = None):  
        ...  
  
    def create_tb(self, conn: connection = None):  
        ...  
  
    ...  

    @staticmethod  
    def print_class_desc(): # 静态方法第一个参数不用填写cls/slef  
        '''  
        静态方法，打印这个类的功能  
        '''  
                TbInit.class_attraibute # 通过类本身访问类属性  
                TbInit.class_method() # 通过类本身访问类方法  
  
        print('TbInit是用于....')  

class DataMove:  
         
    ...  

    def init_copy_tb(self):  
        '''  
        初始化用于储存迁移数据的表  

        '''  

        tb_init_ins = TbInit(self.db, self.sc, self.tb) # 实例化，为后续使用"准备数据表"功能做准备  
        is_tb_ready = tb_init_ins.check_tb() # 检查迁移表是否满足要求  
        if not is_tb_ready: # 不满足  
            tb_init_ins.drop_tb() # 删除迁移表  
            tb_init_ins.create_tb() # 按要求创建迁移表  
    ... 

# 继续用食物的例子(代码形式)解释什么是继承/多态

class StuffedDough:
    '''
    一个塞着馅料的面团罢了
    '''
 
    ingredient_type = '有机物' # 类属性

    def __init__(self, dough: str, stuffing: str): # 实例化方法
        '''
        实例化，获取面团和馅料的具体材料
        '''
        self.dough = dough # 实例属性
        self.stuffing = stuffing

    def cook(self): # 有一个烹饪方法
        ...

    def shape(self): # 有一个塑形方法
        print('面团包着馅料罢了') 

class SteamBun(StuffedDough): # 继承
    '''
    包子
    '''

    def cook(self): # 重写方法
        print('蒸')

    # def shape(self):
    #     print(f'我已经是{self.stuffing}包子的形状了') 

    def dip(self): # 新增方法
        print(f'沾点醋')


class Tortellini(StuffedDough):
    '''
    意式饺子
    '''

    def cook(self): # 重写且和包子不一样，是为"多态"
        print('煮')

    def shape(self):
        print(f'我已经是{self.stuffing}意式饺子的形状了') 

pork_bun = SteamBun('低筋小麦面粉', '猪肉') # 猪肉包子
pork_bun.shape()
pork_bun.dip()

rice_bun = SteamBun('低筋小麦面粉', '大米') # 警示我们，类实例化可供定制的内容需视情况而定
rice_bun.shape() # 当然，如果接受一个大米馅的包子也行。。。

class DataMove:  
      
    ...  
    def init_copy_tb(self):  
        '''  
        初始化用于储存迁移数据的表  
  
        '''  
        tb_init_ins = TbInit(self.db, self.sc, self.tb) # 实例化，为后续使用"准备数据表"功能做准备  
        is_tb_ready = tb_init_ins.check_tb() # 检查迁移表是否满足要求  
        if not is_tb_ready: # 不满足  
            tb_init_ins.drop_tb() # 删除迁移表  
            tb_init_ins.create_tb() # 按要求创建迁移表  

...  
from data_save_with_class import DataMove, DataSave, TbInit  
  
...  
# 数据更新/插入  
uni_key = ['id', 'time'] # 数据的唯一键，具体按实际情况构造，例如通过用户id和行为类型构建数据id间  
data_sv_ins = DataSave(uni_key)  
exist_uni_key = data_sv_ins.get_old_data(data_final) # 查看哪些数据是已经存在，稍后进行更新  
data_to_update, data_to_insert = data_sv_ins.split_update_insert_data(data_final, exist_uni_key) # 将数据分为用于插入的和更新的  
data_sv_ins.update_data(data_to_update) # 更新数据  
data_sv_ins.insert_data(data_to_insert) # 插入数据

class SubTbInit(TbInit): # 继承TbInit  
  
    def check_tb(self): # 改写check_tb  
        ... # 改写的逻辑  
  
sub_init_ins = SubTbInit(...) # 使用SubTbInit