querying1.py

import cProfile
from io import StringIO
import pstats
import contextlib
import time
 
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Date, cast
from sqlalchemy.orm import sessionmaker
 
 
@contextlib.contextmanager
def profiled():
    pr = cProfile.Profile()
    pr.enable()
    yield
    pr.disable()
    s = StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
    ps.print_stats()
    # uncomment this to see who's calling what
    # ps.print_callers()
    print(s.getvalue())
 
 
user = 'root'
pwd = ''
host = 'localhost'
db = 'testing'
 
Base = declarative_base()
 
class Transactions(Base):
    __tablename__ = 'transactions'
 
    id = Column(Integer, primary_key=True)
    name = Column(String)
    description = Column(String)
    country_name = Column(String)
    city_name = Column(String)
    cost = Column(String)
    currency = Column(String)
    created_at = Column(DateTime)
    billing_type = Column(String)
    language = Column(String)
    operating_system = Column(String)
 
    def __repr__(self):
        """How the class Transactions is shown.
 
        :returns: return the string with the id and the name.
 
        """
        return "Transaction(id=%s, name=%s)" % (self.id, self.name)
 
 
def transactions_filtered_by_querying():
    with profiled():
        mysql_engine = create_engine('mysql://{0}:{1}@{2}/{3}'.format(user, pwd, host, db))
        Session = sessionmaker(bind=mysql_engine)
        session = Session()
 
        transactions = (
            session.query(Transactions)
            .filter(
                cast(Transactions.created_at, Date) == '2007-09-02'
            )
            .all()
        )
        return [(t.id, t.name, t.billing_type) for t in transactions]


transactions_filtered_by_querying()