麦子学院 2016-10-18 13:41
Django学习之数据库的查询
回复:0 查看:2413
web应用需要经常向数据库查询相关的数据，
Django语言
也不例外。了解Django中的数据查询操作，对于优化应用的性能，有着重要的意义。
基础信息
Django中，model通过Manager获取QuerySet，每个model至少有objects这个 Manager 。
QuerySet 可以有一个或多个 filter , filter 根据传入的参数返回 QuerySet 结果。
用SQL来对比， QuerySet 相当于 SELECT ， filter 相当于 WHERE 或者是 LIMIT
每次添加一条filter，会获得一个新的 QuerySet 。
思路
利用Django的测试框架，构造测试数据，查看Django构建的SQL，以及查询到的内容。
设计表结构
schema
创建model
创建测试app，并加入settings.py APPS中
python manage.py startapp learning
INSTALLED_APPS = [
...
'learning',
...
]
创建对应的model
class Student(models.Model):
name = models.CharField(max_length=255, help_text="学生姓名")
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
def __str__(self):
return self.name
class Teacher(models.Model):
name = models.CharField(max_length=255, help_text="老师姓名")
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
def __str__(self):
return self.name
class Classe(models.Model):
name = models.CharField(max_length=255, help_text="班级名称")
teacher = models.ForeignKey(Teacher, help_text="老师")
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
def __str__(self):
return self.name
class ClasseStudent(models.Model):
student = models.ForeignKey(Student)
classe = models.ForeignKey(Classe)
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
测试数据
使用factory-boy来辅助构造测试数据，代码如下
class TeacherFactory(factory.DjangoModelFactory):
class Meta:
model = Teacher
name = factory.Sequence(lambda n: "Teacher #%s" % n)
class StudentFactory(factory.DjangoModelFactory):
class Meta:
model = Student
name = factory.Sequence(lambda n: "Student #%s" % n)
class ClasseFactory(factory.DjangoModelFactory):
class Meta:
model = Classe
name = factory.Sequence(lambda n: "Class #%s" % n)
class ClasseStudentFactory(factory.DjangoModelFactory):
class Meta:
model = ClasseStudent
student = factory.SubFactory(StudentFactory)
classe = factory.SubFactory(ClasseFactory)
现在来伪造测试数据
在tests.py添加测试用例
class ModelTestCase(TestCase):
def setUp(self):
self.teacher_one = TeacherFactory()
self.class_one = ClasseFactory(teacher=self.teacher_one)
self.class_two = ClasseFactory(teacher=self.teacher_one)
for i in range(40):
ClasseStudentFactory(student=StudentFactory(), classe=self.class_one)
for i in range(40):
ClasseStudentFactory(student=StudentFactory(), classe=self.class_two)
可以看到，添加了一个老师，两个班，每个班加入了40个学生
filter和exclude
添加testcase
def test_filter_chain(self):
query_set = Student.objects.filter(name__startswith='Student').exclude(pk=1).filter(create_time__year__gte=2015)[3:10]
print query_set.query
print query_set
执行
python manage.py test python manage.py test learning.test.test.ModelTestCase.test_filter_chain
可以看到结果
SELECT `learning_student`.`id`, `learning_student`.`name`, `learning_student`.`create_time` FROM `learning_student` WHERE (`learning_student`.`name` LIKE BINARY Student% AND NOT (`learning_student`.`id` = 1) AND `learning_student`.`create_time` >= 2014-12-31 16:00:00) LIMIT 7 OFFSET 3
<QuerySet [<Student: Student #4>, <Student: Student #5>, <Student: Student #6>, <Student: Student #7>, <Student: Student #8>, <Student: Student #9>, <Student: Student #10>]>
和我们预期的一致，值得注意的是，Django执行的是lazy query，也就是说前面的filter和exclude并没有去查询数据库，后面的分片才导致了那次的查询
order_by
def test_order_by(self):
query_set = Classe.objects.order_by("-create_time")
print query_set.query
print query_set
SELECT `learning_classe`.`id`, `learning_classe`.`name`, `learning_classe`.`teacher_id`, `learning_classe`.`create_time` FROM `learning_classe` ORDER BY `learning_classe`.`create_time` DESC
<QuerySet [<Classe: Class #3>, <Classe: Class #2>]>
如果有多个column参与排序，可以使用 Coalesce
select_related
可以用select_related查询外键的信息，并将结果保存，这样查找外键信息时，将不会向数据库发送请求，如下所示
def test_select_related(self):
query_set = Classe.objects.select_related("teacher")
print query_set.query
print query_set[0].teacher
SELECT `learning_classe`.`id`, `learning_classe`.`name`, `learning_classe`.`teacher_id`, `learning_classe`.`create_time`, `learning_teacher`.`id`, `learning_teacher`.`name`, `learning_teacher`.`create_time` FROM `learning_classe` INNER JOIN `learning_teacher` ON (`learning_classe`.`teacher_id` = `learning_teacher`.`id`)
<QuerySet [<Classe: Class #0>, <Classe: Class #1>]>
defer和only
有时候查询只需要部分字段的结果，可以用defer和only来限制查询的结果
def test_defer(self):
print Classe.objects.defer("create_time").query
print Classe.objects.only("create_time").query
SELECT `learning_classe`.`id`, `learning_classe`.`name`, `learning_classe`.`teacher_id` FROM `learning_classe`
SELECT `learning_classe`.`id`, `learning_classe`.`create_time` FROM `learning_classe`
如果访问到没有获取的字段时，会再从数据库中读一次
文章来源：简书
基础信息
Django中，model通过Manager获取QuerySet，每个model至少有objects这个 Manager 。
QuerySet 可以有一个或多个 filter , filter 根据传入的参数返回 QuerySet 结果。
用SQL来对比， QuerySet 相当于 SELECT ， filter 相当于 WHERE 或者是 LIMIT
每次添加一条filter，会获得一个新的 QuerySet 。
思路
利用Django的测试框架，构造测试数据，查看Django构建的SQL，以及查询到的内容。
设计表结构
schema
创建model
创建测试app，并加入settings.py APPS中
python manage.py startapp learning
INSTALLED_APPS = [
...
'learning',
...
]
创建对应的model
class Student(models.Model):
name = models.CharField(max_length=255, help_text="学生姓名")
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
def __str__(self):
return self.name
class Teacher(models.Model):
name = models.CharField(max_length=255, help_text="老师姓名")
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
def __str__(self):
return self.name
class Classe(models.Model):
name = models.CharField(max_length=255, help_text="班级名称")
teacher = models.ForeignKey(Teacher, help_text="老师")
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
def __str__(self):
return self.name
class ClasseStudent(models.Model):
student = models.ForeignKey(Student)
classe = models.ForeignKey(Classe)
create_time = models.DateTimeField(auto_now_add=True, help_text="创建时间")
测试数据
使用factory-boy来辅助构造测试数据，代码如下
class TeacherFactory(factory.DjangoModelFactory):
class Meta:
model = Teacher
name = factory.Sequence(lambda n: "Teacher #%s" % n)
class StudentFactory(factory.DjangoModelFactory):
class Meta:
model = Student
name = factory.Sequence(lambda n: "Student #%s" % n)
class ClasseFactory(factory.DjangoModelFactory):
class Meta:
model = Classe
name = factory.Sequence(lambda n: "Class #%s" % n)
class ClasseStudentFactory(factory.DjangoModelFactory):
class Meta:
model = ClasseStudent
student = factory.SubFactory(StudentFactory)
classe = factory.SubFactory(ClasseFactory)
现在来伪造测试数据
在tests.py添加测试用例
class ModelTestCase(TestCase):
def setUp(self):
self.teacher_one = TeacherFactory()
self.class_one = ClasseFactory(teacher=self.teacher_one)
self.class_two = ClasseFactory(teacher=self.teacher_one)
for i in range(40):
ClasseStudentFactory(student=StudentFactory(), classe=self.class_one)
for i in range(40):
ClasseStudentFactory(student=StudentFactory(), classe=self.class_two)
可以看到，添加了一个老师，两个班，每个班加入了40个学生
filter和exclude
添加testcase
def test_filter_chain(self):
query_set = Student.objects.filter(name__startswith='Student').exclude(pk=1).filter(create_time__year__gte=2015)[3:10]
print query_set.query
print query_set
执行
python manage.py test python manage.py test learning.test.test.ModelTestCase.test_filter_chain
可以看到结果
SELECT `learning_student`.`id`, `learning_student`.`name`, `learning_student`.`create_time` FROM `learning_student` WHERE (`learning_student`.`name` LIKE BINARY Student% AND NOT (`learning_student`.`id` = 1) AND `learning_student`.`create_time` >= 2014-12-31 16:00:00) LIMIT 7 OFFSET 3
<QuerySet [<Student: Student #4>, <Student: Student #5>, <Student: Student #6>, <Student: Student #7>, <Student: Student #8>, <Student: Student #9>, <Student: Student #10>]>
和我们预期的一致，值得注意的是，Django执行的是lazy query，也就是说前面的filter和exclude并没有去查询数据库，后面的分片才导致了那次的查询
order_by
def test_order_by(self):
query_set = Classe.objects.order_by("-create_time")
print query_set.query
print query_set
SELECT `learning_classe`.`id`, `learning_classe`.`name`, `learning_classe`.`teacher_id`, `learning_classe`.`create_time` FROM `learning_classe` ORDER BY `learning_classe`.`create_time` DESC
<QuerySet [<Classe: Class #3>, <Classe: Class #2>]>
如果有多个column参与排序，可以使用 Coalesce
select_related
可以用select_related查询外键的信息，并将结果保存，这样查找外键信息时，将不会向数据库发送请求，如下所示
def test_select_related(self):
query_set = Classe.objects.select_related("teacher")
print query_set.query
print query_set[0].teacher
SELECT `learning_classe`.`id`, `learning_classe`.`name`, `learning_classe`.`teacher_id`, `learning_classe`.`create_time`, `learning_teacher`.`id`, `learning_teacher`.`name`, `learning_teacher`.`create_time` FROM `learning_classe` INNER JOIN `learning_teacher` ON (`learning_classe`.`teacher_id` = `learning_teacher`.`id`)
<QuerySet [<Classe: Class #0>, <Classe: Class #1>]>
defer和only
有时候查询只需要部分字段的结果，可以用defer和only来限制查询的结果
def test_defer(self):
print Classe.objects.defer("create_time").query
print Classe.objects.only("create_time").query
SELECT `learning_classe`.`id`, `learning_classe`.`name`, `learning_classe`.`teacher_id` FROM `learning_classe`
SELECT `learning_classe`.`id`, `learning_classe`.`create_time` FROM `learning_classe`
如果访问到没有获取的字段时，会再从数据库中读一次
文章来源：简书