0. Pipvenv虚拟环境

1. 安装pipenv

pip install pipenv

2. 查看本地python的版本

where python

3.进入到项目文件夹，使用指定版本的python创建虚拟环境

pipenv shell --python C:\Python312\python.exe

4. 该文件夹执行pip list，可以看出来他是一个纯净的环境

image.png

5.安装dbt

pipenv install dbt-snowflake

1.魔法方法

1.1__getitem__允许类有索引访问方式，自定义自己的字典

索引和赋值方法，用于支持对象的索引和赋值操作。

class MyDict:
    def __init__(self):
        self.data = {}
    
    def __getitem__(self, key):
        return self.data[key]

    def __setitem__(self, key, value):
        self.data[key] = value

my_dict = MyDict()
my_dict['key'] = 'value'
print(my_dict['key'])  # 输出: value

2. OOP

2.1 依赖反转

• 概念：设计面向对象的时候，高层的模块不应该是依赖低层模块， 两者都应该依赖于抽象，意思是：不管是设计这个类，还是去具体实现这个类的时候，都应该设计成为抽象的类。
• 实际流程：（1）设计一个日志接口的抽象类；（2）设计一个文件日志和数据库日志类，继承①的抽象类；（3）创建一个专门用来使用②日志的类，供用户使用（依赖反转体现）；

1. 设计抽象的日志元类

class Logger(ABC):
    @abstractmethod
    def log(self, message):
        pass

    @abstractmethod
    def set_log_level(self, level):
        pass

    @abstractmethod
    def clear_log(self):
        pass

2. 设计具体的日志记录器，文件日志记录和数据库日志记录器等

class FileLogger(Logger):
    def __init__(self):
        self.log_level = 'INFO'  # 默认日志级别为 INFO
    
    def log(self, message):
        with open('logfile.txt', 'a') as f:
            f.write(message + '\n')

    def set_log_level(self, level):
        self.log_level = level

    def clear_log(self):
        open('logfile.txt', 'w').close()  # 清空日志文件

class DatabaseLogger(Logger):
    def log(self, message):
        # 实现将日志消息写入数据库的逻辑
        pass

3. 设计负责使用日志记录器的文件

class LogManager:
    def __init__(self, logger):
        self.logger = logger

    def do_something_and_log(self, action): #传入的是类
        # 做一些操作
        result = action()
        self.logger.log('Action done: ' + result)

    def set_log_level(self, level):
        self.logger.set_log_level(level)

    def clear_log(self):
        self.logger.clear_log()

4. 使用LogManager实现具体功能，而不是直接使用log类

file_logger = FileLogger()
log_manager = LogManager(file_logger)
log_manager.set_log_level('DEBUG')  # 设置日志级别为 DEBUG
log_manager.do_something_and_log(lambda: 'Result')
log_manager.clear_log()  # 清除日志