1、参数查询示例demo代码
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
源代码：https://github.com/mongonauta/heydjbot/tree/master/demo3/server/database.py
2、创建Dataframe示例demo代码
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
源代码：https://github.com/bmassman/fake_news/blob/master/fake_news/pipeline/build_df.py
3、表连接查询示例demo代码
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)
4、分组查询示例demo代码
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
5、导出数据示例demo代码
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]
源代码：https://github.com/EmmaOnThursday/next-book/blob/master/app/recommendation_creation.py
6、数据库连接示例demo代码
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)
7、有关时间类型的示例demo代码
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)
8、时间类型处理的示例demo代码
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)