Python-сообщество

from django.db import models
 
 
class Book(models.Model):
    author = models.ForeignKey('Author')
    title = models.CharField(max_length=100)
    page_count = models.IntegerField()
    exists = models.BooleanField()
    chars_count = models.BigIntegerField()
    digit_version = models.BinaryField()
    important_pages = models.CommaSeparatedIntegerField(max_length=1000)
    wrote = models.DateField(auto_now=True)
    published = models.DateTimeField(auto_now=True)
    cost = models.DecimalField(max_digits=65, decimal_places=2)  # 65 - максимум для MySql
    sale_cost = models.FloatField()
    image = models.ImageField(upload_to='media')
    book_addition = models.FileField(upload_to='media')
 
    class Meta:
        ordering = ['author']
 
 
class Author(models.Model):
    name = models.CharField(max_length=100)
    mail = models.EmailField(max_length=254)
 
 
class Publisher(models.Model):
    name = models.CharField(max_length=100)
    book = models.ManyToManyField('Book')

import sys
import os
import pymysql
from pprint import pprint
from datetime import datetime
from io import StringIO
 
 
pymysql.install_as_MySQLdb()
sys.path.append(r'D:\SCRIPTS\WORK\test_django_db_backends')
os.environ['DJANGO_SETTINGS_MODULE'] = 'test_django_db_backends.settings'
from app_main.models import Author, Book, Publisher
from django import db
from django.core.management import call_command
 
def test_bulk_create(db_engine_name):
    objs_list = []
    test_result = {'db_engine_name': db_engine_name}
    file = open(db_engine_name+'_sql', 'a')
    db.reset_queries()
 
    for i in range(RECORDS_COUNT):
        obj = Author(name='author_'+str(i), mail='author_'+str(i)+'@djangomail.de')
        objs_list.append(obj)
    t1 = datetime.now()
    Author.objects.using(db_engine_name).bulk_create(objs_list)
    t2 = datetime.now()
    authors_bulk_create = str((t2-t1).total_seconds())
    test_result['authors_bulk_create'] = authors_bulk_create
 
    pprint('==========authors_bulk_create==========', file)
    pprint(db.connections[db_engine_name].queries, file)
    db.reset_queries()
 
    objs_list.clear()
    authors = Author.objects.using(db_engine_name).all()
    for i in range(RECORDS_COUNT):
        obj = Book(author=authors[i], title='book_'+str(i), page_count=300, exists=True, chars_count=300000, cost=347.77, sale_cost=244.07)
        objs_list.append(obj)
    t1 = datetime.now()
    Book.objects.using(db_engine_name).bulk_create(objs_list)
    t2 = datetime.now()
    books_bulk_create = str((t2-t1).total_seconds())
    test_result['books_bulk_create'] = books_bulk_create
 
    pprint('==========books_bulk_create==========', file)
    pprint(db.connections[db_engine_name].queries, file)
    db.reset_queries()
 
    objs_list.clear()
    for i in range(RECORDS_COUNT):
        obj = Publisher(name='publisher_'+str(i))
        objs_list.append(obj)
    t1 = datetime.now()
    Publisher.objects.using(db_engine_name).bulk_create(objs_list)
    t2 = datetime.now()
    publisher_bulk_create = str((t2-t1).total_seconds())
    test_result['publisher_bulk_create'] = publisher_bulk_create
 
    pprint('==========publisher_bulk_create==========', file)
    pprint(db.connections[db_engine_name].queries, file)
    db.reset_queries()
 
    t1 = datetime.now()
    for p, b in zip(Publisher.objects.using(db_engine_name).iterator(), Book.objects.using(db_engine_name).iterator()):
        p.book.add(b)
    t2 = datetime.now()
    publisher_m2m_add = str((t2-t1).total_seconds())
    test_result['publisher_m2m_add'] = publisher_m2m_add
 
    pprint('==========publisher_m2m_add==========', file)
    pprint(db.connections[db_engine_name].queries, file)
    file.close()
 
    return test_result
 
def clean_db():
    Author.objects.using('mysql').all().delete()
    Book.objects.using('mysql').all().delete()
    Publisher.objects.using('mysql').all().delete()
    Author.objects.using('postgres').all().delete()
    Book.objects.using('postgres').all().delete()
    Publisher.objects.using('postgres').all().delete()
 
    cursor = db.connection.cursor()
    sql_commands = StringIO()
    call_command('sqlsequencereset', 'app_main', database='postgres', stdout=sql_commands)
    cursor.execute(sql_commands.getvalue())
 
def main():
    clean_db()
    all_testes = [test_bulk_create]
  
    for test in all_testes:
        print('='*50)
        print(test.__name__)
        test_result = test('mysql')
        pprint(test_result)
        test_result = test('postgres')
        pprint(test_result)
 
 
if __name__ == '__main__':
    main()

==================================================
test_bulk_create
{'authors_bulk_create': '0.262104',
 'books_bulk_create': '1.656163',
 'db_engine_name': 'mysql',
 'publisher_bulk_create': '0.816326',
 'publisher_m2m_add': '210.869314'}
{'authors_bulk_create': '0.435174',
 'books_bulk_create': '2.467486',
 'db_engine_name': 'postgres',
 'publisher_bulk_create': '0.196079',
 'publisher_m2m_add': '21.941125'}

db.connections[db_engine_name].queries