六.文件操作

1.read 方法

# 读取文件 - read()
f = open("/Users/qinyingjie/Downloads/改签合同系统操作指引.pdf", "r", encoding="UTF-8")
print(type(f))

# 指定字节读取文件
print(f"读取10个字节的结果：{f.read(10)}")
print(f"read方法读取全部内容的结果是：{f.read()}")

2.readLines 所有行

# 读取文件 - readLines()
f = open("/Users/qinyingjie/Downloads/改签合同系统操作指引.pdf", "r", encoding="UTF-8")
lines = f.readlines()   # 读取文件的全部行，封装到列表中
print(f"lines对象的类型：{type(lines)}")
print(f"lines对象的内容是：{lines}")

3.readline

# 读取文件 - readline()
 line1 = f.readline()
 line2 = f.readline()
 line3 = f.readline()
 print(f"第一行数据是：{line1}")
 print(f"第二行数据是：{line2}")
 print(f"第三行数据是：{line3}")

4.循环读取文件内容

# for循环读取文件行
row = 0
with open("/Users/qinyingjie/Downloads/资料/第1-12章资料/资料/地图数据/疫情.txt", "r", encoding="UTF-8") as f:
    for line in f:
        row += 1
        print(f"第{row}行数据是：{line}")
    f.close()
time.sleep(500000)

5.文件写入

# 打开文件，不存在的文件, r, w, a
import time
f = open("D:/test.txt", "w", encoding="UTF-8")
# write写入
f.write("Hello World!!!")       # 内容写入到内存中
# flush刷新
f.flush()                       # 将内存中积攒的内容，写入到硬盘的文件中

# close关闭
f.close()                       # close方法，内置了flush的功能的

# 打开一个存在的文件
f = open("D:/test.txt", "w", encoding="UTF-8")
# write写入、flush刷新
f.write("黑马程序员")
# close关闭
f.close()

6.追加写入

# 打开文件
f = open("D:/test.txt", "a", encoding="UTF-8")
# write写入
f.write("黑马程序员")
# flush刷新
f.flush()
# close关闭
f.close()

7.文件备份

"""
文件备份
"""

# 打开文件得到文件对象，准备读取
fr = open("D:/bill.txt", "r", encoding="UTF-8")
# 打开文件得到文件对象，准备写入
fw = open("D:/bill.txt.bak", "w", encoding="UTF-8")
# for循环读取文件
for line in fr:
    line = line.strip()
    # 判断内容，将满足的内容写出
    if line.split(",")[4] == "测试":
        continue        # continue进入下一次循环，这一次后面的内容就跳过了
    # 将内容写出去
    fw.write(line)
    # 由于前面对内容进行了strip()的操作，所以要手动的写出换行符
    fw.write("\n")

# close2个文件对象
fr.close()
fw.close()      # 写出文件调用close()会自动flush()

七.异常-模块-包

1.捕获异常

# 基本捕获语法
 try:
     f = open("D:/abc.txt", "r", encoding="UTF-8")
 except:
     print("出现异常了，因为文件不存在，我将open的模式，改为w模式去打开")
     f = open("D:/abc.txt", "w", encoding="UTF-8")

2.捕获指定的异常

# 捕获指定的异常
 try:
     print(name)
 except NameError as e:
     print("出现了变量未定义的异常")
     print(e)

3.捕获多个异常

# 捕获多个异常
 try:
     # 1 / 0
     print(name)
 except (NameError, ZeroDivisionError) as e:
     print("出现了变量未定义 或者 除以0的异常错误")

4.捕获所有异常

# 捕获所有异常
try:
    f = open("D:/123.txt", "r", encoding="UTF-8")
except Exception as e:
    print("出现异常了")
    f = open("D:/123.txt", "w", encoding="UTF-8")
else:
    print("好高兴，没有异常。")
finally:
    print("我是finally，有没有异常我都要执行")
    f.close()

5.异常的传递

# 定义一个出现异常的方法
def func1():
    print("func1 开始执行")
    num = 1 / 0     # 肯定有异常，除以0的异常
    print("func1 结束执行")
# 定义一个无异常的方法，调用上面的方法

def func2():
    print("func2 开始执行")
    func1()
    print("func2 结束执行")
# 定义一个方法，调用上面的方法

def main():
    try:
        func2()
    except Exception as e:
        print(f"出现异常了，异常的信息是：{e}")

main()

6.模块的导入

from time import sleep

# 使用import导入time模块使用sleep功能（函数）
 import time     # 导入Python内置的time模块（time.py这个代码文件）
 print("你好")
 time.sleep(5)   # 通过. 就可以使用模块内部的全部功能（类、函数、变量）
 print("我好")

7.导入全部功能

# 使用 * 导入time模块的全部功能
 from time import *      # *表示全部的意思
 print("你好")
 sleep(5)
 print("我好")

8.添加别名

# 使用as给特定功能加上别名
 import time as t
 print("你好")
 t.sleep(5)
 print("我好")

# 方法别名
from time import sleep as sl
print("你好")
sl(5)
print("我好")

9.导入包

# 创建一个包
# 导入自定义的包中的模块，并使用
import my_package.my_module1
import my_package.my_module2

my_package.my_module1.info_print1()
my_package.my_module2.info_print2()

from my_package import my_module1
from my_package import my_module2
my_module1.info_print1()
my_module2.info_print2()

from my_package.my_module1 import info_print1
from my_package.my_module2 import info_print2
info_print1()
info_print2()

# 通过__all__变量，控制import *
from my_package import *
my_module1.info_print1()
my_module2.info_print2()

10.日期格式化

import time
# time模块
ts = time.time()        # 当前时间戳
print(f"当前时间戳是：{ts}")

# 获取当前时间以指定的格式显示,2000-01-01 10:00:00
print(time.strftime("%Y-%m-%d %H:%M:%S"))

# 将指定的时间戳转换为格式化的日期字符串
print(time.strftime("%Y-%m-%d %H:%M:%S"))

八.对象

1.init 方法

class Record:
    def __init__(self, date, order_id, money, province):
        self.date = date  # 订单日期
        self.order_id = order_id  # 订单ID
        self.money = money  # 订单金额
        self.province = province  # 销售省份

2.str 方法

class Record:
    def __init__(self, date, order_id, money, province):
        self.date = date  # 订单日期
        self.order_id = order_id  # 订单ID
        self.money = money  # 订单金额
        self.province = province  # 销售省份

    def __str__(self):
        return f"{self.date}, {self.order_id}, {self.money}, {self.province}"

3.call 函数

在 Python 中，__call__是一个特殊的方法，允许将一个类的实例像函数一样进行调用。当使用instance()的形式调用一个对象时，__call__方法会被自动调用。

下面是一个简单的示例，展示了如何使用__call__方法：

class MyClass:
    def __init__(self):
        pass

    def __call__(self, *args, **kwargs):
        print("Calling MyClass instance")

# 创建一个MyClass的实例
my_obj = MyClass()

# 调用实例，会自动调用__call__方法
my_obj()

运行以上代码，会输出："Calling MyClass instance"。

__call__方法可以接受任意数量的参数，并且可以通过*args和**kwargs来接收位置参数和关键字参数。在__call__方法内部，您可以根据需要执行与实例调用相关的操作。

通过使用__call__方法，您可以使一个类的实例表现得像一个函数，并且具有函数一样的调用方式和行为。这在某些情况下可以提供更加灵活和方便的代码组织和调用方式。

4.继承

class TextFileReader(FileReader):

class TextFileReader(FileReader):

    def __init__(self, path):
        self.path = path  # 定义成员变量记录文件的路径

    # 复写（实现抽象方法）父类的方法
    def read_data(self) -> list[Record]:
        f = open(self.path, "r", encoding="UTF-8")

        record_list: list[Record] = []
        for line in f.readlines():
            line = line.strip()  # 消除读取到的每一行数据中的\n
            data_list = line.split(",")
            record = Record(data_list[0], data_list[1], int(data_list[2]), data_list[3])
            record_list.append(record)

        f.close()
        return record_list

5.返回值类型

class TextFileReader(FileReader):

    def __init__(self, path):
        self.path = path  # 定义成员变量记录文件的路径

    # 复写（实现抽象方法）父类的方法
    def read_data(self) -> list[Record]:
        f = open(self.path, "r", encoding="UTF-8")

        record_list: list[Record] = []
        for line in f.readlines():
            line = line.strip()  # 消除读取到的每一行数据中的\n
            data_list = line.split(",")
            record = Record(data_list[0], data_list[1], int(data_list[2]), data_list[3])
            record_list.append(record)

        f.close()
        return record_list

6.成员方法

# 定义一个带有成员方法的类
class Student:
    name = None  # 学生的姓名

    def say_hi(self):
        print(f"大家好呀，我是{self.name}，欢迎大家多多关照")

    def say_hi2(self, msg):
        print(f"大家好，我是：{self.name}，{msg}")


stu = Student()
stu.name = "周杰轮"
stu.say_hi2("哎哟不错哟")

stu2 = Student()
stu2.name = "林俊节"
stu2.say_hi2("小伙子我看好你")

7.类和对象

# 设计一个闹钟类
class Clock:
    id = None       # 序列化
    price = None    # 价格


    def ring(self):
        import winsound
        winsound.Beep(2000, 3000)

# 构建2个闹钟对象并让其工作
clock1 = Clock()
clock1.id = "003032"
clock1.price = 19.99
print(f"闹钟ID：{clock1.id}，价格：{clock1.price}")
# clock1.ring()

clock2 = Clock()
clock2.id = "003033"
clock2.price = 21.99
print(f"闹钟ID：{clock2.id}，价格：{clock2.price}")
clock2.ring()

8.构造方法

class Student:

    def __init__(self, name, age ,tel):
        self.name = name
        self.age = age
        self.tel = tel
        print("Student类创建了一个类对象")

stu = Student("周杰轮", 31, "18500006666")
print(stu.name)
print(stu.age)
print(stu.tel)

9.内置方法

class Student:
    def __init__(self, name, age):
        self.name = name        # 学生姓名
        self.age = age          # 学生年龄

    # __str__魔术方法
    def __str__(self):
        return f"Student类对象，name:{self.name}, age:{self.age}"

    # __lt__魔术方法
    def __lt__(self, other):
        return self.age < other.age

    # __le__魔术方法
    def __le__(self, other):
        return self.age <= other.age

    # __eq__魔术方法
    def __eq__(self, other):
        return self.age == other.age


stu1 = Student("周杰轮", 31)
stu2 = Student("林俊节", 36)
print(stu1 == stu2)

10.封装

# 定义一个类，内含私有成员变量和私有成员方法
class Phone:
    __current_voltage = 0.5        # 当前手机运行电压


    def __keep_single_core(self):
        print("让CPU以单核模式运行")


    def call_by_5g(self):
        if self.__current_voltage >= 1:
            print("5g通话已开启")
        else:
            self.__keep_single_core()
            print("电量不足，无法使用5g通话，并已设置为单核运行进行省电。")


phone = Phone()
phone.call_by_5g()

11.多继承

# 演示单继承
class Phone:
    IMEI = None  # 序列号
    producer = "ITCAST"  # 厂商

    def call_by_4g(self):
        print("4g通话")


class Phone2022(Phone):
    face_id = "10001"  # 面部识别ID

    def call_by_5g(self):
        print("2022年新功能：5g通话")


phone = Phone2022()
print(phone.producer)
phone.call_by_4g()
phone.call_by_5g()


# 演示多继承
class NFCReader:
    nfc_type = "第五代"
    producer = "HM"

    def read_card(self):
        print("NFC读卡")

    def write_card(self):
        print("NFC写卡")


class RemoteControl:
    rc_type = "红外遥控"

    def control(self):
        print("红外遥控开启了")


class MyPhone(Phone, NFCReader, RemoteControl):
    pass


phone = MyPhone()
phone.call_by_4g()
phone.read_card()
phone.write_card()
phone.control()
print(phone.producer)

12.复写方法

class Phone:
    IMEI = None  # 序列号
    producer = "ITCAST"  # 厂商

    def call_by_5g(self):
        print("使用5g网络进行通话")


# 定义子类，复写父类成员
class MyPhone(Phone):
    producer = "ITHEIMA"  # 复写父类的成员属性

    def call_by_5g(self):
        print("开启CPU单核模式，确保通话的时候省电")
        # 方式1
        # print(f"父类的厂商是：{Phone.producer}")
        # Phone.call_by_5g(self)
        # 方式2
        print(f"父类的厂商是：{super().producer}")
        super().call_by_5g()
        print("关闭CPU单核模式，确保性能")


phone = MyPhone()
phone.call_by_5g()
print(phone.producer)

# 在子类中，调用父类成员

13.类型注解

# 基础数据类型注解
import json
import random

# var_1: int = 10
# var_2: str = "itheima"
# var_3: bool = True
# 类对象类型注解
class Student:
    pass
stu: Student = Student()

# 基础容器类型注解
# my_list: list = [1, 2, 3]
# my_tuple: tuple = (1, 2, 3)
# my_dict: dict = {"itheima": 666}
# 容器类型详细注解
my_list: list[int] = [1, 2, 3]
my_tuple: tuple[int, str, bool] = (1, "itheima", True)
my_dict: dict[str, int] = {"itheima": 666}
# 在注释中进行类型注解
var_1 = random.randint(1, 10)   # type: int
var_2 = json.loads('{"name": "zhangsan"}')  # type: dict[str, str]
def func():
    return 10
var_3 = func()  # type: int
# 类型注解的限制
var_4: int = "itheima"
var_5: str = 123

14.类型注解

# 对形参进行类型注解
def add(x: int, y: int):
    return x + y

# 对返回值进行类型注解
def func(data: list) -> list:
    return data

print(func(1))

15.Union 类型

# 使用Union类型，必须先导包
from typing import Union

my_list: list[Union[int, str]] = [1, 2, "itheima", "itcast"]

def func(data: Union[int, str]) -> Union[int, str]:
    pass

16.多态

class Animal:
    def speak(self):
        pass


class Dog(Animal):
    def speak(self):
        print("汪汪汪")


class Cat(Animal):
    def speak(self):
        print("喵喵喵")


def make_noise(animal: Animal):
    """制造点噪音，需要传入Animal对象"""
    animal.speak()


# 演示多态，使用2个子类对象来调用函数
dog = Dog()
cat = Cat()

make_noise(dog)
make_noise(cat)


# 演示抽象类
class AC:
    def cool_wind(self):
        """制冷"""
        pass

    def hot_wind(self):
        """制热"""
        pass

    def swing_l_r(self):
        """左右摆风"""
        pass


class Midea_AC(AC):
    def cool_wind(self):
        print("美的空调制冷")

    def hot_wind(self):
        print("美的空调制热")

    def swing_l_r(self):
        print("美的空调左右摆风")


class GREE_AC(AC):
    def cool_wind(self):
        print("格力空调制冷")

    def hot_wind(self):
        print("格力空调制热")

    def swing_l_r(self):
        print("格力空调左右摆风")


def make_cool(ac: AC):
    ac.cool_wind()


midea_ac = Midea_AC()
gree_ac = GREE_AC()

make_cool(midea_ac)
make_cool(gree_ac)

17.类的定义

下面是一个示例的 NewSpaceGraph 类的定义，该类表示一个新空间图（NewSpaceGraph）：

class NewSpaceGraph:
    def __init__(self):
        self.nodes = []
        self.edges = []

    def add_node(self, node):
        self.nodes.append(node)

    def add_edge(self, edge):
        self.edges.append(edge)

    def get_nodes(self):
        return self.nodes

    def get_edges(self):
        return self.edges
class Node:
    def __init__(self, name):
        self.name = name

class Edge:
    def __init__(self, source, target):
        self.source = source
        self.target = target

# 创建NewSpaceGraph对象
graph = NewSpaceGraph()

# 创建节点对象
node1 = Node("A")
node2 = Node("B")
node3 = Node("C")

# 添加节点到图中
graph.add_node(node1)
graph.add_node(node2)
graph.add_node(node3)

# 创建边对象
edge1 = Edge(node1, node2)
edge2 = Edge(node2, node3)

# 添加边到图中
graph.add_edge(edge1)
graph.add_edge(edge2)

# 获取节点列表和边列表
nodes = graph.get_nodes()
edges = graph.get_edges()

# 打印节点列表
print("Nodes:")
for node in nodes:
    print(node.name)

# 打印边列表
print("Edges:")
for edge in edges:
    print(edge.source.name, "->", edge.target.name)

九.SQL

1.自动提交

from pymysql import Connection

# 构建到MySQL数据库的链接
conn = Connection(
    host="localhost",   # 主机名（IP）
    port=3306,          # 端口
    user="root",        # 账户
    password="123456",  # 密码
    autocommit=True     # 设置自动提交
)

# print(conn.get_server_info())
# 执行非查询性质SQL
cursor = conn.cursor()      # 获取到游标对象
# 选择数据库
conn.select_db("world")
# 执行sql
cursor.execute("insert into student values(10001, '周杰轮', 31, '男')")
# 关闭链接
conn.close()

2.数据插入

from pymysql import Connection

# 构建到MySQL数据库的链接
conn = Connection(
    host="localhost",   # 主机名（IP）
    port=3306,          # 端口
    user="root",        # 账户
    password="123456",  # 密码
    autocommit=True     # 自动提交（确认）
)

# print(conn.get_server_info())
# 执行非查询性质SQL
cursor = conn.cursor()      # 获取到游标对象
# 选择数据库
conn.select_db("world")
# 执行sql
cursor.execute("insert into student values(10002, '林俊节', 31, '男')")
# # 通过commit确认
# conn.commit()
# 关闭链接
conn.close()

3.转 json 串

class Record:

    def __init__(self, date, order_id, money, province):
        self.date = date            # 订单日期
        self.order_id = order_id    # 订单ID
        self.money = money          # 订单金额
        self.province = province    # 销售省份


    def __str__(self):
        return f"{self.date}, {self.order_id}, {self.money}, {self.province}"


    def to_json(self):
        d = {"date": self.date, "order_id": self.order_id, "money": self.money, "province": self.province}
        import json
        return json.dumps(d)

4.读取数据插入 sql

from file_define import TextFileReader, JsonFileReader
from data_define import Record
from pymysql import Connection

text_file_reader = TextFileReader("D:/2011年1月销售数据.txt")
json_file_reader = JsonFileReader("D:/2011年2月销售数据JSON.txt")

jan_data: list[Record] = text_file_reader.read_data()
feb_data: list[Record] = json_file_reader.read_data()
# 将2个月份的数据合并为1个list来存储
all_data: list[Record] = jan_data + feb_data

# 构建MySQL链接对象
conn = Connection(
    host="localhost",
    port=3306,
    user="root",
    password="123456",
    autocommit=True
)
# 获得游标对象
cursor = conn.cursor()
# 选择数据库
conn.select_db("py_sql")
# 组织SQL语句
for record in all_data:
    sql = f"insert into orders(order_date, order_id, money, province) " \
          f"values('{record.date}', '{record.order_id}', {record.money}, '{record.province}')"
    # 执行SQL语句
    cursor.execute(sql)

# 关闭MySQL链接对象
conn.close()

5.读取数据插入文件

f = open("d:/output.json", "w", encoding="UTF-8")
# 构建MySQL链接对象
conn = Connection(
    host="localhost",
    port=3306,
    user="root",
    password="123456",
    autocommit=True
)
# 获得游标对象
cursor = conn.cursor()
# 选择数据库
conn.select_db("py_sql")
# 查询
cursor.execute("SELECT * FROM orders")
result = cursor.fetchall()
for r in result:
    record = Record(r[0], r[1], r[2], r[3])
    f.write(record.to_json())
    f.write("\n")

# 关闭MySQL链接对象
conn.close()
f.close()

6.连接池

问题:

对于数据库连接，一般不建议使用全局变量，在每次操作完成后即关闭连接。这是因为长时间保持数据库连接会对性能和资源消耗产生负面影响。与此相反，使用数据库连接池来维护和分配数据库连接是更好的做法。

好处:

连接池的优点是可以在多个线程或进程之间共享，并且可以有效地管理连接数，而无需手动打开和关闭连接。

常用包:

SQLAlchemy 中的 QueuePool 和 DBUtils 中的 PooledDB 都是流行的 Python 数据库连接池实现，它们具有相似的功能但具有一些区别。

QueuePool 是 SQLAlchemy 内置的一个连接池实现，它可以管理一个连接队列，确保每个连接在使用后被适当地关闭。该池使用 Python 自带的 queue 模块实现，并支持可配置的最大连接数、预处理语句等特性。优点是易于使用，无需其他依赖，并与 SQLAlchemy 之间无缝集成。

PooledDB 是 DBUtils 库提供的一个连接池实现，可以与 SQLAlchemy 或其他 Python 数据库库一起使用。它支持多种类型的连接池，并使用 threading 模块实现线程安全，具有更高的性能和稳定性。该库还提供了一些方便的功能，例如自动回收空闲连接等。

十.运算符

1./

与之相对的，/ 表示浮点除法运算，即对两个数进行除法运算后得到一个浮点数结果。例如：

7 / 2 # 输出结果为 3.5

在这个例子中，7 / 2 的值为 3.5，因为 7 除以 2 的结果是 3.5，直接得到一个浮点数结果。

2.//

在 Python 中，// 表示整数除法运算，即对两个数进行除法运算后向下取整。例如：

7 // 2 # 输出结果为 3

在这个例子中，7 // 2 的值为 3，因为 7 除以 2 的结果是 3.5，向下取整后为 3。

3.^

异或运算是一种位运算，它的特点如下：

1. 相同数字异或为 0，不同数字异或为 1。例如，1 ^ 1 = 0，0 ^ 1 = 1。
2. 异或运算满足交换律和结合律。也就是说，a ^ b = b ^ a，(a ^ b) ^ c = a ^ (b ^ c)。
3. 异或运算可以用于交换两个变量的值，而不需要使用中间变量。例如，a = a ^ b; b = a ^ b; a = a ^ b; 就可以交换变量 a 和 b 的值。
4. 将一个数和 0 进行异或运算，结果是这个数本身。例如，a ^ 0 = a。

异或运算常用于编程中的一些场景，例如：

• 在数组中查找只出现一次的数字；
• 交换两个变量的值；
• 将数列中的数按照奇偶性分为两部分。

异或运算的实现非常高效，可以用于大量数据的位运算，而且具有一些独特的应用场景。

交换变量:

以下是使用异或运算交换两个变量的值的 Python 代码示例：

a = 5
b = 7

a = a ^ b
b = a ^ b
a = a ^ b

print("a =", a)  # 输出 a = 7
print("b =", b)  # 输出 b = 5

在上面的代码中，首先将变量 a 和 b 分别赋值为 5 和 7。然后，通过使用异或运算符 ^ 来交换这两个变量的值，具体步骤如下：

1. 让 a = a ^ b，此时 a 的值为 a 和 b 的异或结果，即 a ^ b；
2. 让 b = a ^ b，此时 b 的值为 a ^ b 和 b 的异或结果，即 a；
3. 让 a = a ^ b，此时 a 的值为 a ^ b 和 a 的异或结果，即 b。

4.&

在 Python 中，& 是按位与运算符，用于对两个整数的每个二进制位执行逻辑 AND 操作。当两个二进制数相应位都为 1 时，结果为 1，否则为 0。例如，表达式 3 & 5 的结果为 1，因为 3 的二进制表示为 11，5 的二进制表示为 101，按位与操作结果为 001。

此外，在 Python 中，& 运算符还可以用于 set 类型求交集。例如，当 a 和 b 为两个 set 类型变量时，a & b 将返回 a 和 b 的交集。

以下是一些示例代码：

# 按位与运算
a = 3   # 二进制表示为 11
b = 5   # 二进制表示为 101
c = a & b   # 二进制表示为 001，即十进制的 1
print(c)

# set 类型求交集
set1 = {1, 2, 3, 4, 5}
set2 = {3, 4, 5, 6, 7}
intersection = set1 & set2
print(intersection)

上述代码将输出：

1
{3, 4, 5}

注意，使用 & 运算符求 set 类型的交集时，需要将结果转化为 set 类型或其他类型（例如 list）再进行后续操作。

十一.高级技巧

1.闭包

# 简单闭包
# def outer(logo):
#
#     def inner(msg):
#         print(f"<{logo}>{msg}<{logo}>")
#
#     return inner
#
#
# fn1 = outer("黑马程序员")
# fn1("大家好")
# fn1("大家好")
#
# fn2 = outer("传智教育")
# fn2("大家好")
# 使用nonlocal关键字修改外部函数的值
# def outer(num1):
#
#     def inner(num2):
#         nonlocal num1
#         num1 += num2
#         print(num1)
#
#     return inner
#
# fn = outer(10)
# fn(10)
# fn(10)
# fn(10)
# fn(10)

# 使用闭包实现ATM小案例
def account_create(initial_amount=0):

    def atm(num, deposit=True):
        nonlocal initial_amount
        if deposit:
            initial_amount += num
            print(f"存款：+{num}， 账户余额：{initial_amount}")
        else:
            initial_amount -= num
            print(f"取款：-{num}， 账户余额：{initial_amount}")

    return atm

atm = account_create()

atm(100)
atm(200)
atm(100, deposit=False)

2.装饰器

# 装饰器的一般写法（闭包）
# def outer(func):
#     def inner():
#         print("我睡觉了")
#         func()
#         print("我起床了")
#
#     return inner
#
#
# def sleep():
#     import random
#     import time
#     print("睡眠中......")
#     time.sleep(random.randint(1, 5))
#
#
# fn = outer(sleep)
# fn()

# 装饰器的快捷写法（语法糖）
def outer(func):
    def inner():
        print("我睡觉了")
        func()
        print("我起床了")

    return inner


@outer
def sleep():
    import random
    import time
    print("睡眠中......")
    time.sleep(random.randint(1, 5))


sleep()

3.单例模式

class StrTools:
    pass

str_tool = StrTools()

from str_tools_py import str_tool

s1 = str_tool
s2 = str_tool

print(id(s1))
print(id(s2))

4.工厂模式

class Person:
    pass

class Worker(Person):
    pass

class Student(Person):
    pass

class Teacher(Person):
    pass


class PersonFactory:
    def get_person(self, p_type):
        if p_type == 'w':
            return Worker()
        elif p_type == 's':
            return Student()
        else:
            return Teacher()


pf = PersonFactory()
worker = pf.get_person('w')
stu = pf.get_person('s')
teacher = pf.get_person('t')

5.多线程

import time
import threading


def sing(msg):
    print(msg)
    time.sleep(1)


def dance(msg):
    print(msg)
    time.sleep(1)


if __name__ == '__main__':
    # 创建一个唱歌的线程
    sing_thread = threading.Thread(target=sing, args=("我要唱歌 哈哈哈",))
    # 创建一个跳舞的线程
    dance_thread = threading.Thread(target=dance, kwargs={"msg": "我在跳舞哦 啦啦啦"})

    # 让线程去干活吧
    sing_thread.start()
    dance_thread.start()

6.socket

socket服务端

import socket
# 创建Socket对象
socket_server = socket.socket()
# 绑定ip地址和端口
socket_server.bind(("localhost", 8888))
# 监听端口
socket_server.listen(1)
# listen方法内接受一个整数传参数，表示接受的链接数量
# 等待客户端链接
# result: tuple = socket_server.accept()
# conn = result[0]        # 客户端和服务端的链接对象
# address = result[1]     # 客户端的地址信息
conn, address = socket_server.accept()
# accept方法返回的是二元元组(链接对象， 客户端地址信息)
# 可以通过 变量1, 变量2 = socket_server.accept()的形式，直接接受二元元组内的两个元素
# accept()方法，是阻塞的方法，等待客户端的链接，如果没有链接，就卡在这一行不向下执行了

print(f"接收到了客户端的链接，客户端的信息是：{address}")

while True:
    # 接受客户端信息，要使用客户端和服务端的本次链接对象，而非socket_server对象
    data: str = conn.recv(1024).decode("UTF-8")
    # recv接受的参数是缓冲区大小，一般给1024即可
    # recv方法的返回值是一个字节数组也就是bytes对象，不是字符串，可以通过decode方法通过UTF-8编码，将字节数组转换为字符串对象
    print(f"客户端发来的消息是：{data}")
    # 发送回复消息
    msg = input("请输入你要和客户端回复的消息：")
    if msg == 'exit':
        break
    conn.send(msg.encode("UTF-8"))
# 关闭链接
conn.close()
socket_server.close()

socket客户端

import socket
# 创建socket对象
socket_client = socket.socket()
# 连接到服务端
socket_client.connect(("localhost", 8888))

while True:
    # 发送消息
    msg = input("请输入要给服务端发送的消息：")
    if msg == 'exit':
        break
    socket_client.send(msg.encode("UTF-8"))
    # 接收返回消息
    recv_data = socket_client.recv(1024)        # 1024是缓冲区的大小，一般1024即可。 同样recv方法是阻塞的
    print(f"服务端回复的消息是：{recv_data.decode('UTF-8')}")
# 关闭链接
socket_client.close(

7.正则表达式

import re

s = "1python itheima python python"
# match 从头匹配
result = re.match("python", s)
print(result)
# print(result.span())
# print(result.group())
# search 搜索匹配

result = re.search("python2", s)
print(result)
# findall 搜索全部匹配
result = re.findall("python", s)
print(result)

import re

# s = "itheima1 @@python2 !!666 ##itccast3"
#
# result = re.findall(r'[b-eF-Z3-9]', s)   # 字符串前面带上r的标记，表示字符串中转义字符无效，就是普通字符的意思
# print(result)

# 匹配账号，只能由字母和数字组成，长度限制6到10位
# r = '^[0-9a-zA-Z]{6,10}$'
# s = '123456_'
# print(re.findall(r, s))
# 匹配QQ号，要求纯数字，长度5-11，第一位不为0
# r = '^[1-9][0-9]{4,10}$'
# s = '123453678'
# print(re.findall(r, s))
# 匹配邮箱地址，只允许qq、163、gmail这三种邮箱地址
# abc.efg.daw@qq.com.cn.eu.qq.aa.cc
# abc@qq.com
# {内容}.{内容}.{内容}.{内容}.{内容}.{内容}.{内容}.{内容}@{内容}.{内容}.{内容}
r = r'(^[\w-]+(\.[\w-]+)*@(qq|163|gmail)(\.[\w-]+)+$)'
# s = 'a.b.c.d.e.f.g@qq.com.a.z.c.d.e'
s = 'a.b.c.d.e.f.g@126.com.a.z.c.d.e'
print(re.match(r, s))

8.递归遍历文件

import os


def test_os():
    """演示os模块的3个基础方法"""
    print(os.listdir("D:/test"))        # 列出路径下的内容
    # print(os.path.isdir("D:/test/a"))   # 判断指定路径是不是文件夹
    # print(os.path.exists("D:/test"))    # 判断指定路径是否存在


def get_files_recursion_from_dir(path):
    """
    从指定的文件夹中使用递归的方式，获取全部的文件列表
    :param path: 被判断的文件夹
    :return: list，包含全部的文件，如果目录不存在或者无文件就返回一个空list
    """
    print(f"当前判断的文件夹是：{path}")
    file_list = []
    if os.path.exists(path):
        for f in os.listdir(path):
            new_path = path + "/" + f
            if os.path.isdir(new_path):
                # 进入到这里，表明这个目录是文件夹不是文件
                file_list += get_files_recursion_from_dir(new_path)
            else:
               file_list.append(new_path)
    else:
        print(f"指定的目录{path}，不存在")
        return []

    return file_list

if __name__ == '__main__':
    print(get_files_recursion_from_dir("D:/test"))


def a():
    a()

9.测试用例

import unittest
from binary_tree_template import TreeNode, traverse, search, insert

class TestBinaryTreeTemplate(unittest.TestCase):
    def setUp(self):
        # 构造二叉树
        self.root = TreeNode(5)
        self.root.left = TreeNode(3)
        self.root.right = TreeNode(6)
        self.root.left.left = TreeNode(2)
        self.root.left.right = TreeNode(4)
        self.root.right.right = TreeNode(7)

    def test_traverse(self):
        # 测试遍历框架
        result = []

        def visit(node: TreeNode):
            result.append(node.val)

        traverse(self.root, visit)
        self.assertEqual(result, [2, 3, 4, 5, 6, 7])

    def test_search(self):
        # 测试搜索框架
        node = search(self.root, 4)
        self.assertEqual(node.val, 4)

        node = search(self.root, 8)
        self.assertIsNone(node)

    def test_insert(self):
        # 测试插入框架
        insert(self.root, 1)
        insert(self.root, 8)
        insert(self.root, 9)
        result = []

        def visit(node: TreeNode):
            result.append(node.val)

        traverse(self.root, visit)
        self.assertEqual(result, [1, 2, 3, 4, 5, 6, 7, 8, 9])

if __name__ == '__main__':
    unittest.main()

10.异步

Python 中的异步编程是通过 asyncio 模块来实现的。asyncio 是一个基于事件循环的异步 IO 框架，它提供了协程（coroutine）来实现异步编程。

协程是一种轻量级的线程，它可以在同一个线程中并发执行多个协程，从而实现异步编程。协程可以通过 async 和 await 关键字来定义和调用。

import asyncio

async def async_func():
    print('Start async_func.')
    await asyncio.sleep(1)  # 模拟异步操作
    print('Async_func finished.')

async def main():
    print('Start main.')
    task = asyncio.create_task(async_func())  # 创建协程任务
    await task  # 等待协程任务完成
    print('Main finished.')

asyncio.run(main())  # 运行主协程

在上面的例子中，async_func 是一个异步函数，它通过 await asyncio.sleep(1) 模拟了一个耗时的异步操作。main 函数是一个协程，它通过 asyncio.create_task 创建了一个协程任务，并通过 await 等待任务完成。最后，通过 asyncio.run 运行主协程。

需要注意的是，异步编程需要遵循一些规则，例如不能在协程中使用阻塞 IO 操作，否则会阻塞整个事件循环。因此，需要使用异步 IO 库来替换阻塞 IO 操作。此外，还需要避免死锁和竞态条件等问题，因此需要仔细设计协程的调用顺序和并发控制方式。

11.数据打包

使用元组或列表可以实现打包操作，例如：

a = 10
b = "hello"
c = [1, 2, 3]
d = (a, b, c)
print(d)  # (10, 'hello', [1, 2, 3])

12.数据拆包

使用元组或列表可以实现拆包操作，例如：

d = (10, 'hello', [1, 2, 3])
a, b, c = d
print(a)  # 10
print(b)  # hello
print(c)  # [1, 2, 3]