Python pandas 模块，read_sql_query() 实例源码

def get_classified_songs(self, telegram_id):
        conn = sqlite3.connect(self._DATABASE)
        sql = """
              SELECT
                danceability, energy, loudness, speechiness, acousticness,
                instrumentalness, liveness, valence, tempo, activity
              FROM songs s, users u, song_user su
              WHERE
                activity IS NOT NULL AND
                s.id = su.song_id AND
                su.user_id = u.id AND
                u.telegram_user_id = {}
        """.format(telegram_id)
        resp = pd.read_sql_query(sql, conn)
        conn.close()

        return resp

def build_df(table: str = 'articles',
             start_date: Optional[datetime] = None,
             end_date: Optional[datetime] = None) -> pd.DataFrame:
    """Build dataframe with derived fields."""
    with closing(sqlite3.connect(DB_FILE_NAME)) as conn:
        articles = pd.read_sql_query(f'select * from {table}', conn)
    articles['date'] = pd.to_datetime(articles['publish_date'])
    if start_date:
        articles = articles.loc[articles['date'] >= start_date]
    if end_date:
        articles = articles.loc[articles['date'] <= end_date]

    articles = articles.replace([None], [''], regex=True)
    articles['base_url'] = articles.apply(get_url_base, axis=1)
    articles['word_count'] = articles.apply(count_words, axis=1)
    return articles

def SensitivityQuery(self, table, data_set):
        # Returns the number of times an analyte is found at each concentration and the
        # number of repetitions in a particular data set.
        sql_statement = "SELECT COUNT(%s.id) AS Count, %s.Concentration_pg AS Conc_pg, \
                            DataSetConcentrations.Repetitions AS Repetitions \
                        FROM \
                            Sample \
                        INNER JOIN %s ON \
                            %s.id = Sample.%s_foreignkey \
                        INNER JOIN DataSetConcentrations ON \
                            DataSetConcentrations.id = Sample.DataSetConcentrations_foreignkey \
                        WHERE \
                            Sample.DataSetName = '%s' \
                        GROUP BY \
                            Conc_pg \
                        ORDER BY \
                            Conc_pg;" % (table, table, table, table, table, data_set)
        return pd.read_sql_query(sql_statement, self.conn)

def GetRepsAtEachConcentration(self, analyte_table_lst, data_set):
        df = pd.DataFrame()
        for table in analyte_table_lst:
            sql_statement = "SELECT \
                                %s.Concentration_pg AS Conc, COUNT(%s.Concentration_pg) AS %s \
                            FROM \
                                Sample \
                            Inner Join %s ON \
                                %s.id = Sample.%s_foreignkey \
                            WHERE \
                                DataSetName = '%s' \
                            GROUP BY 1 \
                            ORDER BY 1 ASC;" % (table, table, table, table, table, table, data_set)
            df1 = pd.read_sql_query(sql_statement, self.conn)
            df1.set_index('Conc', inplace=True)

            df = pd.concat([df, df1], axis=1)

        return df

def import_data_from_psql(user_id):
    """Import data from psql; clean & merge dataframes."""
    library = pd.read_sql_table(
        'library',
        con='postgres:///nextbook',
        columns=['book_id', 'title', 'author', 'pub_year', 'original_pub_year', 'pages'])

    book_subjects = pd.read_sql_table(
        'book_subjects',
        con='postgres:///nextbook')

    subjects = pd.read_sql_table(
        'subjects', con='postgres:///nextbook',
        columns=['subject_id', 'subject'])

    user_ratings = pd.read_sql_query(
        sql=('SELECT book_id, user_id, status, rating FROM user_books WHERE user_id=%s' % user_id),
        con='postgres:///nextbook')

    library = library.merge(user_ratings, how='left', on='book_id')
    library['pages'].fillna(0, inplace=True)

    # merge subject names into book_subjects; drop uninteresting subjects from book_subjects table
    book_subjects = book_subjects.merge(subjects, how='left', on='subject_id')
    delete_values = ["protected daisy", "accessible book", "in library", "overdrive", "large type books", 'ficci\xc3\xb3n juvenil', 'ficci\xc3\xb3n', 'lending library']
    book_subjects = book_subjects[~book_subjects['subject'].isin(delete_values)]

    return [library, book_subjects, subjects]

def test_sql_open_close(self):
        # Test if the IO in the database still work if the connection closed
        # between the writing and reading (as in many real situations).

        with tm.ensure_clean() as name:

            conn = self.connect(name)
            sql.to_sql(self.test_frame3, "test_frame3_legacy", conn,
                       flavor="sqlite", index=False)
            conn.close()

            conn = self.connect(name)
            result = sql.read_sql_query("SELECT * FROM test_frame3_legacy;",
                                        conn)
            conn.close()

        tm.assert_frame_equal(self.test_frame3, result)

def test_datetime(self):
        df = DataFrame({'A': date_range('2013-01-01 09:00:00', periods=3),
                        'B': np.arange(3.0)})
        df.to_sql('test_datetime', self.conn)

        # with read_table -> type information from schema used
        result = sql.read_sql_table('test_datetime', self.conn)
        result = result.drop('index', axis=1)
        tm.assert_frame_equal(result, df)

        # with read_sql -> no type information -> sqlite has no native
        result = sql.read_sql_query('SELECT * FROM test_datetime', self.conn)
        result = result.drop('index', axis=1)
        if self.flavor == 'sqlite':
            self.assertTrue(isinstance(result.loc[0, 'A'], string_types))
            result['A'] = to_datetime(result['A'])
            tm.assert_frame_equal(result, df)
        else:
            tm.assert_frame_equal(result, df)

def test_datetime_NaT(self):
        df = DataFrame({'A': date_range('2013-01-01 09:00:00', periods=3),
                        'B': np.arange(3.0)})
        df.loc[1, 'A'] = np.nan
        df.to_sql('test_datetime', self.conn, index=False)

        # with read_table -> type information from schema used
        result = sql.read_sql_table('test_datetime', self.conn)
        tm.assert_frame_equal(result, df)

        # with read_sql -> no type information -> sqlite has no native
        result = sql.read_sql_query('SELECT * FROM test_datetime', self.conn)
        if self.flavor == 'sqlite':
            self.assertTrue(isinstance(result.loc[0, 'A'], string_types))
            result['A'] = to_datetime(result['A'], errors='coerce')
            tm.assert_frame_equal(result, df)
        else:
            tm.assert_frame_equal(result, df)

def test_datetime_time(self):
        # test support for datetime.time
        df = DataFrame([time(9, 0, 0), time(9, 1, 30)], columns=["a"])
        df.to_sql('test_time', self.conn, index=False)
        res = read_sql_table('test_time', self.conn)
        tm.assert_frame_equal(res, df)

        # GH8341
        # first, use the fallback to have the sqlite adapter put in place
        sqlite_conn = TestSQLiteFallback.connect()
        sql.to_sql(df, "test_time2", sqlite_conn, index=False)
        res = sql.read_sql_query("SELECT * FROM test_time2", sqlite_conn)
        ref = df.applymap(lambda _: _.strftime("%H:%M:%S.%f"))
        tm.assert_frame_equal(ref, res)  # check if adapter is in place
        # then test if sqlalchemy is unaffected by the sqlite adapter
        sql.to_sql(df, "test_time3", self.conn, index=False)
        if self.flavor == 'sqlite':
            res = sql.read_sql_query("SELECT * FROM test_time3", self.conn)
            ref = df.applymap(lambda _: _.strftime("%H:%M:%S.%f"))
            tm.assert_frame_equal(ref, res)
        res = sql.read_sql_table("test_time3", self.conn)
        tm.assert_frame_equal(df, res)

def test_connectable_issue_example(self):
        # This tests the example raised in issue
        # https://github.com/pydata/pandas/issues/10104

        def foo(connection):
            query = 'SELECT test_foo_data FROM test_foo_data'
            return sql.read_sql_query(query, con=connection)

        def bar(connection, data):
            data.to_sql(name='test_foo_data',
                        con=connection, if_exists='append')

        def main(connectable):
            with connectable.connect() as conn:
                with conn.begin():
                    foo_data = conn.run_callable(foo)
                    conn.run_callable(bar, foo_data)

        DataFrame({'test_foo_data': [0, 1, 2]}).to_sql(
            'test_foo_data', self.conn)
        main(self.conn)

def test_temporary_table(self):
        test_data = u'Hello, World!'
        expected = DataFrame({'spam': [test_data]})
        Base = declarative.declarative_base()

        class Temporary(Base):
            __tablename__ = 'temp_test'
            __table_args__ = {'prefixes': ['TEMPORARY']}
            id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
            spam = sqlalchemy.Column(sqlalchemy.Unicode(30), nullable=False)

        Session = sa_session.sessionmaker(bind=self.conn)
        session = Session()
        with session.transaction:
            conn = session.connection()
            Temporary.__table__.create(conn)
            session.add(Temporary(spam=test_data))
            session.flush()
            df = sql.read_sql_query(
                sql=sqlalchemy.select([Temporary.spam]),
                con=conn,
            )

        tm.assert_frame_equal(df, expected)

def _get_index_columns(self, tbl_name):
        ixs = sql.read_sql_query(
            "SHOW INDEX IN %s" % tbl_name, self.conn)
        ix_cols = {}
        for ix_name, ix_col in zip(ixs.Key_name, ixs.Column_name):
            if ix_name not in ix_cols:
                ix_cols[ix_name] = []
            ix_cols[ix_name].append(ix_col)
        return list(ix_cols.values())

    # TODO: cruft?
    # def test_to_sql_save_index(self):
    #     self._to_sql_save_index()

    #     for ix_name, ix_col in zip(ixs.Key_name, ixs.Column_name):
    #         if ix_name not in ix_cols:
    #             ix_cols[ix_name] = []
    #         ix_cols[ix_name].append(ix_col)
    #     return ix_cols.values()

def get_data(query):
    """
    Pulls data from the db based on the query

    Input
    -----
    query: str
       SQL query from the database

    Output
    ------
    data: DataFrame
       Dump of Query into a DataFrame
    """
    from setup_environment import db_dict
    with setup_environment.connect_to_syracuse_db(**db_dict) as conn:
        data = pd.read_sql_query(query, conn)
    return data

def show_hexbin(self, query):
        """shows hexbin plot over map

        Args: 
            query: name of sql
        """
        self.load()
        data = pd.read_sql_query(con=self.con, sql=query)
        points = self.gen_points(data, self.data_map)
        hx = self.base_map.hexbin(
            np.array([geom.x for geom in points]),
            np.array([geom.y for geom in points]),
            gridsize=275,
            bins='log',
            mincnt=1,
            edgecolor='none',
            alpha=1.,
            lw=0.2,
            cmap=plt.get_cmap('afmhot'))
        plt.tight_layout()
        plt.show()

def show_scatter(self, query, color='blue'):
        self.load()
        """shows scatter plot over map

        Args: 
            query: name of sql
        """
        data = pd.read_sql_query(con=self.con, sql=query)
        points = self.gen_points(data, self.data_map)

        plot = self.base_map.scatter(
            [point.x for point in points],
            [point.y for point in points],
            10, marker='o', lw=.25,
            facecolor=color, edgecolor='w',
            alpha=0.9, antialiased=True,
            zorder=3)
        plt.show()

def get_classified_songs(self, telegram_id):
        conn = sqlite3.connect(self._DATABASE)
        sql = """
              SELECT
                danceability, energy, loudness, speechiness, acousticness,
                instrumentalness, liveness, valence, tempo, activity
              FROM songs s, users u, song_user su
              WHERE
                activity IS NOT NULL AND
                s.id = su.song_id AND
                su.user_id = u.id AND
                u.telegram_user_id = {}
        """.format(telegram_id)
        resp = pd.read_sql_query(sql, conn)
        conn.close()

        return resp

def get_upstream_stops_ratio(self, target, trough_stops, ratio):
        """
        Selects the stops for which the ratio or higher proportion of trips to the target passes trough a set of trough stops
        :param target: target of trips
        :param trough_stops: stops where the selected trips are passing trough
        :param ratio: threshold for inclusion
        :return:
        """
        if isinstance(trough_stops, list):
            trough_stops = ",".join(trough_stops)
        query = """SELECT stops.* FROM other.stops, 
                    (SELECT q2.from_stop_I AS stop_I FROM 
                    (SELECT journeys.from_stop_I, count(*) AS n_total FROM journeys
                    WHERE journeys.to_stop_I = {target} 
                    GROUP BY from_stop_I) q1,
                    (SELECT journeys.from_stop_I, count(*) AS n_trough FROM journeys, legs 
                    WHERE journeys.journey_id=legs.journey_id AND legs.from_stop_I IN ({trough_stops}) AND journeys.to_stop_I = {target}
                    GROUP BY journeys.from_stop_I) q2
                    WHERE q1.from_stop_I = q2.from_stop_I AND n_trough/(n_total*1.0) >= {ratio}) q1
                    WHERE stops.stop_I = q1.stop_I""".format(target=target, trough_stops=trough_stops, ratio=ratio)
        df = read_sql_query(query, self.conn)
        return df

def get_directly_accessible_stops_within_distance(self, stop, distance):
        """
        Returns stops that are accessible without transfer from the stops that are within a specific walking distance
        :param stop: int
        :param distance: int
        :return:
        """
        query = """SELECT stop.* FROM
                    (SELECT st2.* FROM 
                    (SELECT * FROM stop_distances
                    WHERE from_stop_I = %s) sd,
                    (SELECT * FROM stop_times) st1,
                    (SELECT * FROM stop_times) st2
                    WHERE sd.d < %s AND sd.to_stop_I = st1.stop_I AND st1.trip_I = st2.trip_I 
                    GROUP BY st2.stop_I) sq,
                    (SELECT * FROM stops) stop
                    WHERE sq.stop_I = stop.stop_I""" % (stop, distance)
        return pd.read_sql_query(query, self.conn)

def setUp(self):
        self.data = {
            "DC_PEC": '''
import pandas as pd
from sqlalchemy import create_engine
from urllib.request import urlretrieve; urlretrieve('https://s3.amazonaws.com/assets.datacamp.com/production/course_998/datasets/Chinook.sqlite', 'Chinook.sqlite')
engine = create_engine('sqlite:///Chinook.sqlite')
            ''',
            "DC_CODE": '''
# Execute query and store records in dataframe: df
df = pd.read_sql_query("ELECT * FROM PlaylistTrack INNER JOIN Track on PlaylistTrack.TrackId = Track.TrackId WHERE Milliseconds < 250000", engine)

# Print head of dataframe
print(df.head())
            ''',
            "DC_SOLUTION": '''
# Execute query and store records in dataframe: df
df = pd.read_sql_query("SELECT * FROM PlaylistTrack INNER JOIN Track on PlaylistTrack.TrackId = Track.TrackId WHERE Milliseconds < 250000", engine)

# Print head of dataframe
print(df.head())
'''
        }

def test_Pass(self):
        self.data["DC_SCT"] = '''
# Test: call to read_sql_query() and 'df' variable

test_correct(
        lambda: test_object("df"),
        lambda: test_function("pandas.read_sql_query", do_eval = False)
)

# Test: Predefined code
predef_msg = "You don't have to change any of the predefined code."
test_function("print", incorrect_msg = predef_msg)

success_msg("Great work!")
        '''
        sct_payload = helper.run(self.data)
        self.assertFalse(sct_payload['correct'])

def run(x, string):
    print("Processing chunk: {}".format(string))
    conn = sqlite3.connect(':memory:')
    c = conn.cursor()
    try:
        c.execute("""CREATE TABLE war (nb_trick int)""")
        conn.commit()
    except sqlite3.OperationalError:
        pass

    for i in range(x):
        b = Cython_War_Trick.Battle()
        result = b.trick()
        c.execute("""INSERT INTO war VALUES (?)""", [result])
    conn.commit()

    chunk = pd.read_sql_query("""SELECT nb_trick FROM war""", conn)
    f = chunk['nb_trick'].value_counts()
    return f

def run(self):
        result_list = []
        sql = None
        while True:
            self.lock.acquire()
            if not self.queue.empty():
                sql = self.queue.get()
                self.lock.release()
            else:
                self.lock.release()
                break
            stock_data = pd.read_sql_query(sql, con=self.db)
            stock_data = stock_data.set_index('datetime')
            result_list.append(stock_data)
            print "A stock has finished reading and {} stocks left".format(self.queue.qsize())

        self.lock.acquire()
        self.parent_list.extend(result_list)
        self.lock.release()
        self.db.close()

def get_nr_particles_per_population(self) -> pd.Series:
        """

        Returns
        -------
        nr_particles_per_population: pd.DataFrame
            A pandas DataFrame containing the number
            of particles for each population

        """
        query = (self._session.query(Population.t)
                 .join(ABCSMC)
                 .join(Model)
                 .join(Particle)
                 .filter(ABCSMC.id == self.id))
        df = pd.read_sql_query(query.statement, self._engine)
        nr_particles_per_population = df.t.value_counts().sort_index()
        return nr_particles_per_population

def get_predict_acc2(debug=False):
    db = Db()
    engine = db._get_engine()
    sql_stocklist = "select  * from acc1"
    if debug:
        pass
    df = pd.read_sql_query(sql_stocklist, engine)
    acc2 = df.sort_values('c_yearmonthday', ascending=0)
    acc2 = acc2.head(2)
    acc2 = acc2.groupby('c_yearmonthday').sum()

    acc2_final = pd.DataFrame()
    acc2_final['h_p_acc'] = [df['acc'].sum() / float(df['acc'].count())]
    acc2_final['h_p_change'] = [df['p_change'].sum() / 2.0]
    acc2_final['p_acc'] = [acc2['acc'].sum() / 2.0]
    acc2_final['p_change'] = [acc2['p_change'].sum() / 2.0]

    return acc2_final

def get_data(self,labels_=None, data=None):
        print('Loading CleanText from DataBase from...')
        conn = connect('/home/gondin/metis/project/clinton-email-download/hrcemail3.sqlite')
        sql = """SELECT Keywords, Polarity, Subjectivity, "from", cluster_labels, pdf_path as "Email" FROM document;"""
        self.data = pd.read_sql_query(sql, conn)
        self.data['Similarity'] = self.similarity[:,0] 

        conn.close()

        #self.data = self.data.sample(1000)
        self.data = self.data.sample(15000,random_state=44)


        # labels_ =self.labels_ ==self.label_
        labels_ = self.labels_
        print(self.data.shape)
        print(labels_.shape)
        self.data.Polarity = self.data.Polarity.apply(lambda x: round(x,2)) 



        return (self.data.ix[labels_ & (self.data.cluster_labels>0), ['Keywords','Similarity','Polarity', 'Subjectivity', "from","Email"]].sort_values('Similarity'))
        #return (self.data.ix[labels_, ['Keywords','dist', "Email"]].sort_values('dist'))

def get_data(self,labels_=None, data=None):
        print('Loading CleanText from DataBase from...')
        conn = connect('/home/gondin/metis/project/clinton-email-download/hrcemail3.sqlite')
        sql = """SELECT Keywords, Polarity, Subjectivity, "from", cluster_labels, pdf_path as "Email" FROM document;"""
        self.data = pd.read_sql_query(sql, conn)
        self.data['Similarity'] = self.similarity[:,0] 

        conn.close()

        #self.data = self.data.sample(1000)
        self.data = self.data.sample(15000,random_state=44)


        # labels_ =self.labels_ ==self.label_
        labels_ = self.labels_
        print(self.data.shape)
        print(labels_.shape)
        self.data.Polarity = self.data.Polarity.apply(lambda x: round(x,2)) 



        return (self.data.ix[labels_ & (self.data.cluster_labels>0), ['Keywords','Similarity','Polarity', 'Subjectivity', "from","Email"]].sort_values('Similarity'))
        #return (self.data.ix[labels_, ['Keywords','dist', "Email"]].sort_values('dist'))

def query_wikidata_mysql(query):
    with SSHTunnelForwarder((ssh_host, ssh_port), ssh_username=ssh_user, ssh_pkey=mypkey,
                            remote_bind_address=(sql_hostname, sql_port)) as tunnel:
        conn = pymysql.connect(host='127.0.0.1', user=sql_user, password=sql_pass, db=sql_main_database,
                               port=tunnel.local_bind_port)

        df = pd.read_sql_query(query, conn)
        conn.close()
    return df

def tweet_dates(self):
        conn = sqlite3.connect(self._db, detect_types=sqlite3.PARSE_COLNAMES)
        df = pd.read_sql_query(
            'SELECT created_at FROM tweets', conn, parse_dates=['created_at'],
            index_col=['created_at']
            )
        return df

def all(self):
        conn = sqlite3.connect(self._db, detect_types=sqlite3.PARSE_DECLTYPES)
        df = pd.read_sql_query(
            'SELECT * FROM tweets', conn, parse_dates=['created_at']
            )
        return df

def tweets_since(self, dt):
        """
        Retrieves all tweets since a particular datetime as a generator that
        iterates on ``chunksize``.

        :param dt: The starting datetime to query from.
        """
        conn = sqlite3.connect(self._db, detect_types=sqlite3.PARSE_DECLTYPES)
        df = pd.read_sql_query(
            'SELECT * FROM tweets WHERE created_at > ?', conn, params=(dt,),
            parse_dates=['created_at']
            )
        return TweetBin(df, dt, datetime.now())

def tweets_between(self, start, end):
        """
        Retrieve tweets between the start and and datetimes. Returns a generator
        that iterates on ``chunksize``.

        :param start: The start of the search range.
        :type start: datetime
        :param end: The end of the search range.
        :type end: datetime
        """
        conn = sqlite3.connect(self._db, detect_types=sqlite3.PARSE_DECLTYPES)
        df = pd.read_sql_query(
            'SELECT * FROM tweets WHERE created_at > ? AND created_at <= ?',
            conn, params=(start, end), parse_dates=['created_at']
            )
        return TweetBin(df, start, end)

def get_related_id(session, engine, parameters):
    category = parameters.split(":",1)[0]
    sql_query=session.query(ALLOWED_CLASSES[category])
    for field_value in parameters.split(":",1)[1].split("&&"):
        field, value = field_value.split(".",1)
        if ":" in value:
            values = get_related_id(session, engine, value)
            for value in values:
                value=int(value) # the value is returned as a numpy object
                if field[-4:] == "date": # support for date entry matching (the values have to be passes as string but matched as datetime)
                    value = datetime.datetime(*[int(i) for i in value.split(",")])
                sql_query = sql_query.filter(getattr(ALLOWED_CLASSES[category], field)==value)
        else:
            if field[-4:] == "date": # support for date entry matching (the values have to be passes as string but matched as datetime)
                value = datetime.datetime(*[int(i) for i in value.split(",")])
            sql_query = sql_query.filter(getattr(ALLOWED_CLASSES[category], field)==value)
    mystring = sql_query.statement
    mydf = pd.read_sql_query(mystring,engine)
    mydf = mydf.T.groupby(level=0).first().T #awkward hack to deal with polymorphic tables returning multiple IDs
    related_table_ids = mydf["id"]
    input_values = list(related_table_ids)
    if input_values == []:
        raise BaseException("No entry was found with a value of \""+str(value)+"\" on the \""+field+"\" column of the \""+category+"\" CATEGORY, in the database.")
    session.close()
    engine.dispose()
    return input_values

def get_for_protocolize(db_path, class_name, code):
    """Return a dataframe containing a specific entry from a given class name, joined with its related tables up to three levels down.
    """
    session, engine = load_session(db_path)
    cols = []
    joins = []
    classobject = ALLOWED_CLASSES[class_name]
    insp = sqlalchemy.inspection.inspect(classobject)
    for name, col in insp.columns.items():
        cols.append(col.label(name))
    for name, rel in insp.relationships.items():
        alias = aliased(rel.mapper.class_, name=name)
        joins.append((alias, rel.class_attribute))
        for col_name, col in sqlalchemy.inspection.inspect(rel.mapper).columns.items():
            #the id column causes double entries, as it is mapped once on the parent table (related_table_id) and once on the child table (table_id)
            if col.key != "id":
                aliased_col = getattr(alias, col.key)
                cols.append(aliased_col.label("{}_{}".format(name, col_name)))

        sub_insp = sqlalchemy.inspection.inspect(rel.mapper.class_)
        for sub_name, sub_rel in sub_insp.relationships.items():
            if "contains" not in sub_name:
                sub_alias = aliased(sub_rel.mapper.class_, name=name+"_"+sub_name)
                joins.append((sub_alias, sub_rel.class_attribute))
                for sub_col_name, sub_col in sqlalchemy.inspection.inspect(sub_rel.mapper).columns.items():
                    #the id column causes double entries, as it is mapped once on the parent table (related_table_id) and once on the child table (table_id)
                    if sub_col.key != "id":
                        sub_aliased_col = getattr(sub_alias, sub_col.key)
                        cols.append(sub_aliased_col.label("{}_{}_{}".format(name, sub_name, sub_col_name)))

    sql_query = session.query(*cols).select_from(classobject)
    for join in joins:
        sql_query = sql_query.outerjoin(*join)
    sql_query = sql_query.filter(classobject.code == code)

    mystring = sql_query.statement
    mydf = pd.read_sql_query(mystring,engine)

    session.close()
    engine.dispose()
    return mydf

def DataSetAnalytes(self, data_set, TableProcessingToReturn='Both_PeakFinding_TargetAnalyteFinding'):
        # Query all foreign key columns in Sample table and return a list of all analyts that are
        # found in a given data set.
        column_string = self.createQueryColumnsStr(TableProcessingToReturn)

        # Build SQL statement & Query sample table for all foreign key columns of a given data set
        sql_statement = "SELECT %s FROM Sample WHERE Sample.DataSetName = '%s';" % (column_string, data_set)        
        df = pd.read_sql_query(sql_statement, self.conn)

        return self.GetFoundAnalytesLst(df)

def Get_100fgArea(self, table, data_set):
        sql_statement = "SELECT %s.Area AS Area_100fg \
                        FROM \
                            Sample \
                        Inner Join %s ON \
                            %s.id = Sample.%s_foreignkey \
                        WHERE \
                            Sample.DataSetName = '%s' AND \
                            %s.Concentration_pg = 0.1;"  % (table, table, table, table, data_set, table)

        return pd.read_sql_query(sql_statement, self.conn)

def GetAveSimilarities(self, table, data_set):
        sql_statement = "SELECT AVG(%s.Similarity) AS Ave_Similarity, %s.Concentration_pg AS Conc_pg \
                        FROM \
                            Sample \
                        Inner Join %s ON \
                            %s.id = Sample.%s_foreignkey \
                        WHERE \
                            Sample.DataSetName = '%s' \
                        GROUP BY Conc_pg;"  % (table, table, table, table, table, data_set)

        return pd.read_sql_query(sql_statement, self.conn)

def Similarities(self, table, data_sets):
        # This query provides all the data to create the Concentration vs Similarity plot and tables 
        #condition = "DataSet = '" + "' OR DataSet = '".join(data_sets) + "' "
        condition = self.CreateConditionClause_OrSeriesStr(data_sets)

        sql_statement = "SELECT %s.Similarity AS Similarity, %s.Concentration_pg AS Conc_pg, \
                            Sample.Instrument AS SerNo, Sample.DataSetName AS DataSet \
                        FROM \
                            Sample \
                        Inner Join %s ON \
                            %s.id = Sample.%s_foreignkey \
                        WHERE \
                            %s \
                        ORDER BY SerNo, Conc_pg ASC;"  % (table, table, table, table, table, condition)

        return pd.read_sql_query(sql_statement, self.conn)

def ClearDataSetData(self, data_sets, analyte_table_lst):
        # Clears all the data from the database for the define data sets

        # append the DataSetConcentrations table so it too will be included in the data clearing
        analyte_table_lst.append('DataSetConcentrations')

        # Create a single string that contains all the forigen keys in the sample table
        # that has data to be removed (which is comma space delimited)
        ForeignKeyColumn_lst = [col + '_foreignkey' for col in analyte_table_lst]
        ForeignKeyColumn_Columns = ', '.join(ForeignKeyColumn_lst)

        # Get condition string from data sets
        data_set_condition = self.CreateConditionClause_OrSeriesStr(data_sets, "DataSetName")

        # Resulting df: columns correspond to tables that contain data to be removed, the values in each column
        # are the primary keys of records within that table that need to be deleted.
        sql_statement = 'SELECT %s FROM Sample WHERE %s;' % (ForeignKeyColumn_Columns, data_set_condition)
        df = pd.read_sql_query(sql_statement, self.conn)
        df.columns = analyte_table_lst

        # Iterate through the dataframe by column to delete each record from the db
        # note: column = table name
        for column in df:
            condition = self.CreateConditionClause_OrSeriesStr(set(df[column]), "id")
            self.conn.execute('DELETE FROM %s WHERE %s' % (column, condition))

        # Finally remove the defined records from the sample table as well
        self.conn.execute('DELETE FROM Sample WHERE %s' % data_set_condition)
        self.CommitDB()

def write_locations(field, query_template, plate_name, row, conn, config):
    # Read cells file for each image
    query = query_template.replace("@@@",field).format(
            plate_name,
            row["Metadata_Well"],
            row["Metadata_Site"]
    )
    locations = pd.read_sql_query(query, conn)

    # Keep center coordinates only, remove NaNs, and transform to integers
    locations = locations.dropna(axis=0, how="any")
    locations[field+"_Location_Center_X"] = locations[field+"_Location_Center_X"]*config["compression"]["scaling_factor"]
    locations[field+"_Location_Center_Y"] = locations[field+"_Location_Center_Y"]*config["compression"]["scaling_factor"]
    locations[field+"_Location_Center_X"] = locations[field+"_Location_Center_X"].astype(int)
    locations[field+"_Location_Center_Y"] = locations[field+"_Location_Center_Y"].astype(int)

    # Save the resulting dataset frame in the output directory
    loc_file = "{}/{}/locations/{}-{}-{}.csv".format(
        config["compression"]["output_dir"],
        row["Metadata_Plate"],
        row["Metadata_Well"],
        row["Metadata_Site"],
        field
    )
    dataset.utils.check_path(loc_file)
    locations.to_csv(loc_file, index=False)

def run(self):
        self.initialize()
        tags = pd.read_sql_query("SELECT * FROM TAGS", self.con)
        self.collect_changes(tags)
        self.apply_changes_if_not_dry()
        self.close()

def df_query(self, query, with_labels=False):
        """
        Run a :mod:`sqlalchemy` query and return result as a :class:`pandas.DataFrame`

        Args:

            query (sqlalchemy.orm.query.Query): query object, usually generated by :func:`session.query()` in
                an :class:`sqlalchemy.orm.session.Session`

            with_labels (bool): A query for fields with the same name from different tables will cause problems
                when converting it to a :class:`pandas.DataFrame`, because there will be duplicate column names.
                When setting `with_labels=True`, disambiguation labels are assigned to all (!)
                fields in the query - the field name is prefixed with the column name. This enables
                querying fields with identical names from multiple tables but getting unique column names in the output.

        :return: query result as :class:`pandas.DataFrame`
        """
        import pandas as pd

        if with_labels:
            query = query.with_labels()

        # compile sql statement, including arguments
        statement = query.statement.compile(self.engine)

        # run query
        return pd.read_sql_query(sql=statement, con=self.engine)

def get_pandas_df(self, sql):
        """
        We need to overide this method in order to use connections other than the default_conn_name
        :param sql: A query input via the web UI at /admin/queryview/
        :return: a Panda data frame
        """
        conn = self.get_conn(self.snowflake_conn_id)
        df = pd.read_sql_query(sql, conn)
        return df

def get_test_2(self):
        return pd.read_sql_query('select date,open,close,high,low,vol,code '
                                 'from (select * from t_stick_data_m_test order by date desc limit 15) a '
                                 'order by date asc;',self.sqlconn)

def build_model(all_features, categorical_features, target, connection_string, filename):
    engine = create_engine(connection_string)

    query = """SELECT * FROM connection_features"""
    df = pd.read_sql_query(query, con=engine, index_col=['departurestop', 'departuredate', 'route'])
    df.index.levels[0].name = 'stationfrom'
    df.index.levels[1].name = 'date'
    df.index.levels[2].name = 'vehicle'
    df = df.reset_index()

    model = build_model_random_forest(df, all_features, categorical_features, target)
    joblib.dump(model, filename)

def predict(all_features, categorical_features, connection_string, filename):
    engine = create_engine(connection_string)
    model = joblib.load(filename)
    print model

    query = """SELECT * FROM connection_features"""
    df = pd.read_sql_query(query, con=engine, index_col=['departurestop', 'departuredate', 'route'])
    df.index.levels[0].name = 'stationfrom'
    df.index.levels[1].name = 'date'
    df.index.levels[2].name = 'vehicle'
    df = df.reset_index()

    predicted = model.predict(df[all_features])
    print predicted

def pivot_stations(df, engine):
    query = """
    SELECT
      d.*,
      s.name AS arrivalname
    FROM distance d INNER JOIN station s ON d.stationto = s.id
    """
    distances = pd.read_sql_query(query, con=engine)
    stations = distances['arrivalname'].unique().tolist()
    dist_pivot = pd.pivot_table(distances, values='distance', index=['stationfrom', 'date', 'trip'],
                                columns=['arrivalname'], aggfunc=np.mean)
    dist_pivot = dist_pivot.reindex(df.index.rename(['stationfrom', 'date', 'vehicle']))
    df = df.join(dist_pivot, how='outer')
    return df, stations

def select_fromindexs(dbname,field,indexs,**kwargs):
    normal_db = False
    tablename = False

    # handling even if indexs arent in str format
    if type(indexs[0]) == int:
        indexs = [str(row) for row in indexs]

    for key,value in kwargs.iteritems():
        if key == 'size':
            size = value
        if key == 'normal_db':
            normal_db = value
        if key == 'tablename':
            tablename = value
    a,engine = make_query(dbname,tablename=tablename,normal_db=normal_db)

    stringindexs = ','.join(indexs)

    if not tablename == False:
        dbname = tablename

    # now making querry
    query = '''SELECT * FROM %s WHERE %s IN (%s);''' % (dbname,field,stringindexs)

    return pd.read_sql_query(query,engine)

def select_fromindexs(dbname,field,indexs,**kwargs):
    normal_db = False
    tablename = False

    # handling even if indexs arent in str format
    if type(indexs[0]) == int:
        indexs = [str(row) for row in indexs]

    for key,value in kwargs.iteritems():
        if key == 'size':
            size = value
        if key == 'normal_db':
            normal_db = value
        if key == 'tablename':
            tablename = value
    a,engine = make_query(dbname,tablename=tablename,normal_db=normal_db)

    stringindexs = ','.join(indexs)

    if not tablename == False:
        dbname = tablename

    # now making querry
    query = '''SELECT * FROM %s WHERE %s IN (%s);''' % (dbname,field,stringindexs)

    return pd.read_sql_query(query,engine)

def runCode(self, code):
        errors = None
        output = undefined
        try:
            if code.lstrip().upper().startswith('SELECT '):
                output = pandas.read_sql_query(code, self._connection)
            else:
                self._connection.execute(code)
        except Exception as exc:
            errors = [{
                'line': 0,
                'column': 0,
                'message': str(exc)
            }]

        return {
            'errors': errors,
            'output': None if output is undefined else pack(output)
        }

def query_object(self, starnames, key='*'):
        """
        Get information about the given star.

        Parameters:
        ===========
        starnames:    string, or iterable of strings
                      The name(s) of the star.

        key:          string, default='*' (return everything)
                      What data do you want? Can be anything that sql will take

        Returns:
        ========
        A pandas DataFrame with the given information for each star
        """

        if isinstance(starnames, str):
            starnames = [starnames,]
        starnames = ["'{}'".format(n) for n in starnames]

        name_list = '(' + ', '.join(starnames) + ')'
        sql_query = "SELECT {} FROM star WHERE name IN {}".format(key, name_list)
        print(sql_query)
        df = pd.read_sql_query(sql_query, self.db_con)

        return df