手动连接管理

手动连接管理在每个需要查询数据库的地方都需要打开数据库连接，以下是特别需要注意的点

模型例子

class Log(BaseModel):
    id = AutoField()
    date = DateTimeField(default=datetime.now)
    log = CharField()

class Step(BaseModel):
    id = AutoField()
    name= CharField()
    log = ForeignKeyField(Log, backref='step')

需要查询的场景

外键对象

step.log、step.log.id，但使用 step.log_id 不会去查数据库

backref反向引用

log.step 只会生成查询语句

SELECT `t1`.`id`, `t1`.`name`, `t1`.`log_id` FROM `step` AS `t1` WHERE (`t1`.`log_id` = 1)

再使用其他立即求结果的方法，就会导致连接数据库

log.step[0]
list(log.step[0])
for step in log.step:

不查询的场景

只做 select时不查询数据库

Step.select().where(Step.id==1)

反向引用 backref

from enum import IntEnum, unique
from pathlib import Path
from datetime import datetime
from peewee import *

@unique
class Status(IntEnum):
    CREATED = 0
    FAIL = -1
    SUCCESS = 1


db_path = Path(__file__).parent / "test.sqlite"
database = SqliteDatabase(db_path)

class BaseModel(Model):
    class Meta:
        database = database

class Unit(BaseModel):
    name = TextField(unique=True)
    some_field = TextField(null=True)
    created_at = DateTimeField(default=datetime.now)

class Campaign(BaseModel):
    id_ = AutoField()
    created_at = DateTimeField(default=datetime.now)

class Task(BaseModel):
    id_ = AutoField()
    status = IntegerField(default=Status.CREATED)
    unit = ForeignKeyField(Unit, backref="tasks1")
    campaign = ForeignKeyField(Campaign, backref="tasks")

# 创建数据库连接并创建表格
database.connect()
database.create_tables([Unit, Campaign, Task])

unit = Unit.select().where(Unit.name=="Unit 3").get()


tasks = unit.tasks1
for t in tasks:
    print(t)

使用Enum

默认值

status = IntegerField(default=Status.CREATED)

from enum import IntEnum, unique
from pathlib import Path
from datetime import datetime
from peewee import *

@unique
class Status(IntEnum):
    CREATED = 0
    FAIL = -1
    SUCCESS = 1


db_path = Path(__file__).parent / "test.sqlite"
database = SqliteDatabase(db_path)

class BaseModel(Model):
    class Meta:
        database = database

class Unit(BaseModel):
    name = TextField(unique=True)
    some_field = TextField(null=True)
    created_at = DateTimeField(default=datetime.now)

class Campaign(BaseModel):
    id_ = AutoField()
    created_at = DateTimeField(default=datetime.now)

class Task(BaseModel):
    id_ = AutoField()
    status = IntegerField(default=Status.CREATED)
    unit = ForeignKeyField(Unit, backref="tasks")
    campaign = ForeignKeyField(Campaign, backref="tasks")

# 创建数据库连接并创建表格
database.connect()
database.create_tables([Unit, Campaign, Task])

unit = Unit.create(name="a")
campaign = Campaign.create()

Task.create(unit=unit, campaign=campaign)

开启调试

logger = logging.getLogger("peewee.pool")
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)

问题

当父类中查询到了实例信息，存在某个变量中

在子类中再去使用这些实例关联的外键信息时，不需要查询数据库？

from peewee import *

database = SqliteDatabase(
    "test.db",
    autoconnect=False,
)


class Author(Model):
    id = AutoField()
    name = CharField()

    class Meta:
        database = database


# 定义书籍模型类
class Book(Model):
    id = AutoField()
    title = CharField()
    author = ForeignKeyField(Author, backref="books")

    class Meta:
        database = database


# 创建表格

with database.connection_context():
    database.create_tables([Author, Book], safe=True)
    author = Author.create(name="author1").save()
    book = Book.create(title="book1", author=author).save()


class A:
    def __init__(self):
        with database.connection_context():
            self.book = Book.select().where(Book.title == "book1").first()


class B(A):
    def __init__(self):
        super().__init__()
        # print(0, self.book.author.name)

        with database.connection_context():
            # self._author = self.book.author
            print(self.book.author)  # 这里已经对self.book.author进行查询缓存了，再次取 self.book.author 对象下的属性比如 name 不需要查数据库


class C(B):
    def __init__(self):
        super().__init__()
        print(1, self.book.author.name)


other = C()