bosukh

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime


defined_tables = {}

def dynamic_table_wrapper(func):
    '''
    Decorate a function that declare dynamic table.
    It expects that the function being decorated

bosukh

from sqlalchemy import Column, Integer, String, DateTime

def Trasaction(Base, year, month, day):
    month = "{:02d}".format(month)
    day = "{:02d}".format(day)

    class Trasaction(Base):
        __tablename__ = 'card_trasaction_{year}_{month}_{day}'.format(
            year=year, month=month, day = day
        )

bosukh

# CREATE USER 'example'@'localhost' IDENTIFIED BY 'example_password';
# GRANT ALL PRIVILEGES ON *.* TO 'example2'@'localhost' IDENTIFIED BY 'example_password';
# CREATE DATABASE example;
import sys
import os
import time
from datetime import date, timedelta
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.dialects.mysql import DATETIME, DATE

bosukh

#!/bin/bash

if [ "$#" -ne 5 ]; then
    echo "Wrong number of arguments to the import shell script."
    exit 1
fi

csv_filename=$1
table_name=$2
db_name=$3

bosukh

import os
import time


def test_raw_update(n=100000):
    init_sqlalchemy(False)
    t0 = time.time()
    f = open('customers_to_insert', 'w')
    for i in xrange(1, n+1):
        f.write(

bosukh

import time

def test_mysql_update(n=100000):
    init_sqlalchemy(False)
    t0 = time.time()
    for i in xrange(n):
        DBSession.execute(
            "UPDATE customer SET name = '{0}' WHERE '{1}'".format(
                'NEW_NAME ' + str(i), str(i)
            )

bosukh

import time


def test_sqlalchemy_core_update(n=100000):
    init_sqlalchemy(False)
    t0 = time.time()
    engine.execute(
        Customer.__table__.update(),
        [{'name': 'NEW_NAME ' + str(i), '_id': i}
         for i in xrange(1, n+1)]

bosukh

import time


def test_sqlalchemy_orm_bulk_update(n=100000):
    init_sqlalchemy(False)
    t0 = time.time()
    DBSession.bulk_update_mappings(
        Customer,
        [{'name': 'NEW_NAME ' + str(i), 'id': i}
         for i in xrange(1, n+1)]

bosukh

import time
from sqlalchemy.sql.expression import bindparam


def test_sqlalchemy_orm_update_in_chuck(n=100000):
    init_sqlalchemy(False)
    t0 = time.time()
    customer = Customer.__table__
    stmt = customer.update().where(customer.c.id == bindparam('_id')).values(
        {'name': bindparam('name')}

bosukh

import time

def test_sqlalchemy_orm_update(n=100000):
    init_sqlalchemy(False)
    t0 = time.time()
    customer = Customer.__table__
    for i in xrange(1, n+1):
        stmt = customer.update().where(customer.c.id==i).values(
            {'name': 'NEW_NAME ' + str(i)}
        )