來源：Pexels
筆者在工作中經常要使用SQL，其不乏存在惱人的細微差異和種種限制，但說到底，它是數據行業的基石。因此，對于每一位數據領域的工作者，SQL都是不可或缺的。精通SQL意義非凡。
SQL是很不錯，但怎么能僅滿足于“不錯”呢？為什么不進一步操作SQL呢？
陳述性語句會誘發SQL限制的發生，就是說，向SQL尋求數據，SQL會在特定數據庫找尋并反饋。對于許多數據提取或簡單的數據操作任務來說，這已經足夠了。
但如果有更多需求怎么辦？
本文將為你展示如何操作。
從基礎開始
import pyodbc
from datetime import datetime
classSql:
def__init__(self, database, server="XXVIR00012,55000"):
# here we are telling Python what to connect to (our SQL Server)
self.cnxn = pyodbc.connect("Driver={SQL Server Native Client 11.0};"
"Server="+server+";"
"Database="+database+";"
"Trusted_Connection=yes;")
# initialise query attribute
self.query ="-- {}nn-- Made in Python".format(datetime.now()
.strftime("%d/%m/%Y"))
這個代碼就是操作MS SQL服務器的基礎。只要編寫好這個代碼，通過Python 連接到SQL 僅需:
sql = Sql('database123')
很簡單對么？同時發生了幾件事，下面將對此代碼進行剖析。class Sql:
首先要注意，這個代碼包含在一個類中。筆者發現這是合乎邏輯的，因為在此格式中，已經對此特定數據庫進行了增添或移除進程。若見其工作過程，思路便能更加清晰。
初始化類:
def __init__(self, database,server="XXVIR00012,55000"):
因為筆者和同事幾乎總是連接到相同的服務器，所以筆者將這個通用瀏覽器的名稱設為默認參數server。
在“Connect to Server”對話框或者MS SQL Server Management Studio的視窗頂端可以找到服務器的名稱：
下一步，連接SQL:
self.cnxn =pyodbc.connect("Driver={SQL Server Native Client 11.0};"
"Server="+self.server+";"
"Database="+self.database+";"
"Trusted_Connection=yes;")
pyodbc 模塊，使得這一步驟異常簡單。只需將連接字符串過渡到 pyodbc.connect(...) 函數即可，點擊以了解詳情here。
最后，筆者通常會在 Sql 類中編寫一個查詢字符串，sql類會隨每個傳遞給類的查詢而更新:
self.query = "-- {}nn--Made in Python".format(datetime.now()
.strftime("%d/%m/%Y"))
這樣便于記錄代碼，同時也使輸出更為可讀，讓他人讀起來更舒服。
請注意在下列的代碼片段中，筆者將不再更新代碼中的self.query 部分。
組塊
一些重要函數非常有用，筆者幾乎每天都會使用。這些函數都側重于將數據從數據庫中傳入或傳出。
以下圖文件目錄為始:
對于當前此項目，需要:
· 將文件導入SQL
· 將其合并到單一表格內
· 根據列中類別靈活創建多個表格
SQL類不斷被充實后，后續會容易很多:
import sys
sys.path.insert(0, r'C:\Usermediumpysqlpluslib')
import os
from data importSql
sql =Sql('database123') # initialise the Sql object
directory =r'C:\Usermediumdata\' # this is where our generic data is stored
file_list = os.listdir(directory) # get a list of all files
for file in file_list: # loop to import files to sql
df = pd.read_csv(directory+file) # read file to dataframe
sql.push_dataframe(df, file[:-4])
# now we convert our file_list names into the table names we have imported to SQL
table_names = [x[:-4] for x in file_list]
sql.union(table_names, 'generic_jan') # union our files into one new table called 'generic_jan'
sql.drop(table_names) # drop our original tables as we now have full table
# get list of categories in colX, eg ['hr', 'finance', 'tech', 'c_suite']
sets =list(sql.manual("SELECT colX AS 'category' FROM generic_jan GROUP BY colX", response=True)['category'])
for category in sets:
sql.manual("SELECT * INTO generic_jan_"+category+" FROM generic_jan WHERE colX = '"+category+"'")
從頭開始。
入棧數據結構
defpush_dataframe(self, data, table="raw_data", batchsize=500):
# create execution cursor
cursor = self.cnxn.cursor()
# activate fast execute
cursor.fast_executemany =True
# create create table statement
query ="CREATE TABLE ["+ table +"] (n"
# iterate through each column to be included in create table statement
for i inrange(len(list(data))):
query +="t[{}] varchar(255)".format(list(data)[i]) # add column (everything is varchar for now)
# Append correct connection/end statement code
if i !=len(list(data))-1:
query +=",n"
else:
query +="n);"
cursor.execute(query) # execute the create table statement
self.cnxn.commit() # commit changes
# append query to our SQL code logger
self.query += ("nn-- create tablen"+ query)
# insert the data in batches
query = ("INSERT INTO [{}] ({})n".format(table,
'['+'], [' # get columns
.join(list(data)) +']') +
"VALUESn(?{})".format(", ?"*(len(list(data))-1)))
# insert data into target table in batches of 'batchsize'
for i inrange(0, len(data), batchsize):
if i+batchsize >len(data):
batch = data[i: len(data)].values.tolist()
else:
batch = data[i: i+batchsize].values.tolist()
# execute batch insert
cursor.executemany(query, batch)
# commit insert to SQL Server
self.cnxn.commit()
此函數包含在SQL類中，能輕松將Pandas dataframe插入SQL數據庫。
其在需要上傳大量文件時非常有用。然而，Python能將數據插入到SQL的真正原因在于其靈活性。
要橫跨一打Excel工作簿才能在SQL中插入特定標簽真的很糟心。但有Python在，小菜一碟。如今已經構建起了一個可以使用Python讀取標簽的函數，還能將標簽插入到SQL中。
Manual（函數）
defmanual(self, query, response=False):
cursor = self.cnxn.cursor() # create execution cursor
if response:
returnread_sql(query, self.cnxn) # get sql query output to dataframe
try:
cursor.execute(query) # execute
except pyodbc.ProgrammingErroras error:
print("Warning:n{}".format(error)) # print error as a warning
self.cnxn.commit() # commit query to SQL Server
return"Query complete."
此函數實際上應用在union 和 drop 函數中。僅能使處理SQL代碼變得盡可能簡單。
response參數能將查詢輸出解壓到DataFrame。generic_jan 表中的colX ，可供摘錄所有獨特值，操作如下:
sets =list(sql.manual("SELECT colX AS 'category' FROM generic_jan GROUP BYcolX", response=True)['category'])
Union（函數）
構建 了manual 函數,創建 union 函數就簡單了:
defunion(self, table_list, name="union", join="UNION"):
# initialise the query
query ="SELECT * INTO ["+name+"] FROM (n"
# build the SQL query
query +=f'n{join}n'.join(
[f'SELECT [{x}].* FROM [{x}]'for x in table_list]
)
query +=") x" # add end of query
self.manual(query, fast=True) # fast execute
創建 union 函數只不過是在循環參考 table_list提出的表名，從而為給定的表名構建 UNION函數查詢。然后用self.manual(query)處理。
Drop（函數）
上傳大量表到SQL服務器是可行的。雖然可行，但會使數據庫迅速過載。 為解決這一問題，需要創建一個drop函數:
defdrop(self, tables):
# check if single or list
ifisinstance(tables, str):
# if single string, convert to single item in list for for-loop
tables = [tables]
for table in tables:
# check for pre-existing table and delete if present
query = ("IF OBJECT_ID ('["+table+"]', 'U') IS NOT NULL "
"DROP TABLE ["+table+"]")
self.manual(query) # execute
view rawpysqlplus_drop_short.py hosted with ? by GitHub
點擊
https://gist.github.com/jamescalam/b316c1714c30986fff58c22b00395cc0
得全圖
同樣，此函數也由于 manual 函數極為簡單。操作者可選擇輸入字符到tables ，刪除單個表，或者向tables提供一列表名，刪除多個表。
當這些非常簡單的函數結合在一起時，便可以利用Python的優勢極大豐富SQL的功能。
筆者本人幾乎天天使用此法，其簡單且十分有效。
希望能夠幫助其他用戶找到將Python并入其SQL路徑的方法，感謝閱讀!
