本文主要介绍Python之ETL的前置知识点!肥肠重要

在开始之前聊一下一个点,就是Python中会有很多操作路径的地方(例如文件打开),这里如果单纯从Explore拷贝路径例如

C:\Users\Stray\PycharmProjects\Project

Python是读取不了的,

解决办法

  • 写两个\,就是在\后面加上一个\
  • 路径前面写r”C:\Users\Stray\PycharmProjects\Project“

这样Python就能识别了.

按照重要性依次递减如下PymysqlOS模块UnitTest单元测试Log日志Time模块Recursion递归

Pymsql

pymysql是什么这里就不多bb了,贴上两个链接🔗

github官网

官方文档

Tips

  • 非事务性的SQL是不需要提交的cursor.commit()什么是非事务性的操作, 就是查询select

  • pymysql采用事务方式实现增删改操作,还要手工提交数据

  • cursor.fetchone() 适合获取数据表中指定的某条数据, 格式为元组

  • cursor.fetchall() 获取数据表中的所有数据, 格式为 元组嵌套元组((记录), (记录))

  • cursor.fetchmany(2) 获取前两条数据

  • Q: 连接数据库需要什么

    • A: ① 主机IP ② 端口 ③ 用户名 ④ 密码 ⑤ 数据库 ⑥ 编码格式(简记: h,p,u,p,d,c)

  • Q: 游标是什么

    • A: 相当于数据表中一个指向数据的箭头

Pymysql事务性操作

Demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 导入模块
import pymysql
# 创建连接
conn = pymysql.connect(
    # h p u p d c
    host='localhost',# 可以写成'127.0.0.1'都是本机地址,靠,计网也很重要嗷
    port=3306,
    user='root',
    password='123456',
    database='test',
    charset='utf8'
)
# 获取游标cursor
cursor = conn.cursor()
# 书写sql语句
sql = 'update test.stu set age = 18 where sid = 2;'
# 执行sql语句
row_nums = cursor.execute(sql)
# 向数据库提交执行的结果
conn.commit()
# 判断数据是否提交成功
if row_nums > 0:
    print("数据更新成功!")
else:
    print("数据更新失败!")
# 关闭游标与连接对象
cursor.close()
conn.close()

OS模块

OS模块是Python自带的一个模块,继而还有os.path模块

OS模块的主要作用就是对文件或文件夹进行操作，拥有的功能包括：
① 创建文件夹
② 切换程序目录
③ 获取目录下的文件信息
④ 删除文件、文件夹

Tips:

  • OS模块常用的os.getcwd(),os.listdir(“./“)
  • os.path模块:
    • os.path.abspath
    • os.path.dirname
    • os.path.basename
    • os.path.join
    • os.path.isfile
    • os.path.isdir

Demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import os
# 获取当前工作路径
print(os.getcwd())
# 获取指定目录下(写路径,既可以是绝对路径,也可以是相对路径)的所有文件信息, 返回一个list列表
print(os.listdir("C:\MyBlog"))  # 绝对路径
print(os.listdir("./"))  # 相对路径
# 获取当前文件信息
print(__file__)
# 把相对路径转换为绝对路径
print(os.path.abspath('./'))
# 获取指定位置的上一级路径(括号里面填写路径,也可以写当前文件__file__表示获取当前文件的上一级路径
print(os.path.dirname(__file__))
# 获取指定路径的最后一部分. 比如C:\MyBlog 只会获取到MyBlog,但是一般用的是相对路径
print(os.path.basename("C:\MyBlog"))
# 路径拼接
print(os.path.join('D:\PycharmProjects\python_etl', 'learn\myos'))
# 判断是否为文件(返回值为boolean类型
print(os.path.isfile(__file__))
# 判断是否为路径
print(os.path.isdir(__file__))

UnitTest单元测试

功能介绍

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
"""
unittest模块主要是用来做单元测试
1, 创建一个测试类
2, 书写setUp方法,在测试方法前执行
3, 书写tearDown(扯下,撕下,拆毁)方法,在测试方法后执行
"""
from unittest import TestCase
# 创建测试类(测试类必须继承自TestCase类
class TestDemo(TestCase):
    # -> 提示该方法没有返回值,是一个软性限制,(不遵守会警告,但是不会报错)
    def setUp(self) -> None:
        print('setUp方法被执行!')
    def tearDown(self) -> None:
        print("tearDown方法被执行")
    # 所有的测试方法都要以test开头,否则无法进行单元测试
    def test_func1(self):
        print("测试方法1被执行!")
    def test_func2(self):
        print("测试方法2被执行")

Demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from unittest import TestCase
import logging
class TestLoggingUtil(TestCase):
    # 测试方法,测试的数据可以在这里写, 类似于init方法
    def setUp(self) -> None:
        pass
    def test_get_logger(self):
        """
        这个方法用来测试get_logger函数是否合格,如果这个方法返回一个 日志对象 就证明合格了
        expected: []
        :return: 
        """
        # 这里需要导入要测试的包中的函数 例如 from util.logging_util import logger
        logger = get_logger()
        # 获取原生的logging日志对象类型(通过判断自己写的代码和logging日志对象类型是否想同进行代码测试
        a = logging.RootLogger
        # 当期望值和实际值是同一个类型的时候,测试通过
        self.assertIsInstance(logger, logging.RootLogger)

Log模块

在导入日志模块后, 日志输出到文件中分为4步

  • 1, 创建日志对象: logger = logging.getLogger()
  • 2, 创建日志处理器, 同时设置格式 并绑定处理器到日志对象:
    • file_handler = logging.FileHandler(文件名, 写入方式, 解码)
    • fmt = logging.Formatter(访问时间, 日志级别, 文件名, 哪一行, 具体信息)
  • 3, 设置日志级别 logger.setFormatter(fmt)
  • 4, 输出日志 logger.info()/warning()/error()/critical()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
"""
python中日志操作四步: 输出日志到文件中
"""
# 导入模块
import logging
# 1,创建日志对象
logger = logging.getLogger()
# 2, 创建日志处理器,设置格式并绑定处理器到日志对象
file_handler = logging.FileHandler(
    filename='./test.log',
    mode='a',
    encoding='utf8'
)
# 设置日志格式
# asctime：访问时间
# levelname：日志级别
# filename：具体是在哪个文件中触发的日志
# lineno：具体是这个文件中的哪一行
# message：代表具体输出的日志信息
fmt = logging.Formatter('%(asctime)s - [%(levelname)s] - %(filename)s[%(lineno)d]：%(message)s')
file_handler.setFormatter(fmt)
# 把日志处理器绑定在日志对象
logger.addHandler(file_handler)
# 3, 设置日志级别
logger.setLevel(logging.INFO)
# 4, 输出日志
logger.info('这是一个info级别的日志')
logger.warning('这是一个warning级别的日志')
logger.error('这是一个error级别的日志')
logger.critical('这是一个critical级别的日志')

Time模块

几个重要的点:

  • time.sleep() 休眠(秒)
  • time.time() 获取当前时间戳
  • time.localtime() 获取struct_time对象(本地时间)
  • time.mktime() 将对象转为时间戳
  • time.strftime() 将对象转为指定日期格式
  • time.strptime() 将日期转为对象
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import time
# 程序休眠几秒钟
time.sleep(0)
# 获取当前时间(以秒为单位,time.time_ns()获取当前时间以纳秒为单位
print(time.time())  # 1688021060.9478912
# 获取当前的本地时间, 结果是一个struct_time类的对象
obj = time.localtime()  
print(obj) 
# time.struct_time(tm_year=2023, tm_mon=6, tm_mday=29, tm_hour=14, tm_min=44, tm_sec=20, tm_wday=3, tm_yday=180, tm_isdst=0)
# 将struct_time对象转换为数值类型的时间戳
date = time.mktime(obj)
print(date)  # 1688021060.0
# 将Struct_time类的对象转换为指定格式的日期
date_str1 = time.strftime('%Y-%m-%d %H:%M:%S', obj)
print(date_str1)  # 2023-06-29 14:44:20
# strptime 将日期类型转换为struct_time对象
date3 = time.strptime('2022年11月11日 08时02分24秒', '%Y年%m月%d日 %H时%M分%S秒')
print(date3)  
# time.struct_time(tm_year=2022, tm_mon=11, tm_mday=11, tm_hour=8, tm_min=2, tm_sec=24, tm_wday=4, tm_yday=315, tm_isdst=-1)

Recursion递归

个人理解: 就是函数(方法)自身调用自身的的一个过程(简单说就是套娃), 但是为了不成为死循环,要加上结束条件.

递归的三个必要条件:

  • 在函数体内部调用函数本身
  • 要有明确的递归出口(递归跳出条件)
  • 不能超出最大调用深度(python默认函数最多嵌套1000层)

斐波那契数列Fibonacci

1
2
3
4
5
6
7
8
9
10
# 定义递归函数(1,自身调用自身  2, 有退出条件)
def fibonacci(n):
    # 如果为0或者为1 就给调用者返回1
    if n == 0 or n == 1:
        return 1
    else:
        # 这里用到数学公式n! = n * (n-1)!即为 5! = 5 * 4 * 3 * 2 * 1 = 5 * 4!
        return n * fibonacci(n-1)

print(fibonacci(5))