# Python SQLAlchemy
最近在進行Checkmarx原始碼掃描時出現了一個風險，叫做`Second_Order_SQL_Injection`，詳細的說明就是 : 攻擊者可以直接入侵系統的所有資料庫。 攻擊者可以竊取系統儲存的任何敏感資訊（例如個人用戶詳細訊息或信用卡），並可能更改或刪除現有資料。 (複製貼上Checkmarx的解釋 嘿嘿)
簡單來說，就是如果有一個惡意的使用者透過你提供的輸入欄位寫入了一段不合法的輸入，而你的程式又沒有進行一些檢核或保護，直接組成SQL語言的話，就有可能被攻擊者竊取資料，或是修改資料
舉個例子，假設我們有一個輸入欄位，在使用者輸入並按下「確認」後，就組成一個SQL語法，並傳送到資料庫中執行，如下:
```sql
SELECT userid, password from user where userid='(你輸入的值)'
```
在正常的情況下，我們可以返回一個正常的結果出來，但是如果有一天，有一個攻擊者故意輸入了`';update user set password='1234`
整個字串就會變成
```sql
SELECT userid, password from user where userid='';update user set password='1234'
```
原本提供給使用者的查詢功能瞬間就變成一個更新全表的指令了，因此不管是使用者輸入、讀取檔案，或是讀取資料庫中的資料都要特別小心驗證輸入值，避免攻擊者做出傷害伺服器的事情
## SQLAlchemy
講了這麼多，那這個問題有什麼方法可以解決呢?根據checkMarx的解答，他會希望你將輸入的值全部轉成字串後，將單引號給取代掉，確實這樣就可以解決一些問題了，但是我們還是無法保證輸入的值是否是安全的，攻擊者依然有很多方法可以搞你的資料庫，所以這種情況下，我們就可以使用所謂的ORM物件關聯對映，英文叫 Object Relational Mapping [ORM的解釋](https://zh.wikipedia.org/wiki/%E5%AF%B9%E8%B1%A1%E5%85%B3%E7%B3%BB%E6%98%A0%E5%B0%84)
而目前在Python中，SQLAlchemy是一個可以提供不少關聯式資料庫連接方法的 ORM
要安裝很簡單
```
pip install SQLAlchemy
```
而SQLAlchemy提供了兩種主要的模式
1. SQL表示式語言（SQLAlchemy Core）
2. ORM
而首先我們先介紹連線的方式
### 連線資料庫
在連線的資料庫中，目前我看到的有提供 MySQL、Oracle、PostgreSQL、Microsoft SQL Server、sqllite 等等連線，算是涵蓋了大部分的關聯式資料庫了，在下面我們簡單介紹一下個資料庫如何連線
```python=
from sqlclachemy import create_engine
# PostgreSQL
# 默認情況(即使用psycopg2)
engine = create_engine('postgresql://scott:tiger@localhost/mydatabase')
# MySQL
# 默認情況(即使用mysql-python)
engine = create_engine('mysql://scott:tiger@localhost/foo')
# Oracle
# 默認情況(即使用cx_oracle)
engine = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')
# sqlite 因為是基於文件的資料庫，所以與前面的URL不太一樣
# 在Windows 中使用原始字符串
engine = create_engine(r'sqlite:///C:\path\to\foo.db')
```
而我最常用的Microsoft SQL Server就重點解析一下連線字串如下
```python=
from sqlclachemy import create_engine
#Microsoft SQL Server
# 默認情況(即使用pyodbc)
user = 'willy' # 帳號
pwd = 'willy123' # 密碼
host = '127.0.0.1' # 資料庫主機位置(名稱)
port = '1433' # 連接阜
dbname = 'master' #db名稱
driver='SQL+Server+Native+Client+11.0'
conn_str = 'mssql+pyodbc://{user}:{pwd}@{host}:{port}/{dbname}?driver={driver}'.format(
user=user,
pwd=pwd,
host=host,
port=port,
dbname=dbname,
driver=driver
)
engine = create_engine(conn_str)
```
如此一來就可以連線上資料庫了
### SQL表示式語言（SQLAlchemy Core）
連線上資料庫後，我們就可以試著進行 CRUD 操作了
而上面有提過SQLAlchemy提供了兩種主要的模式，首先我們先介紹第一種模式
#### Define and Create Tables
我們先要定義並且建立一個 table 的結構在程式中，目的是可以讓我們建立的 table 可以映射到真實資料庫中的 table
```python=
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
metadata = MetaData()
# 宣告 table 格式
users = Table('users', metadata,
Column('id', Integer, primary_key=True, autoincrement=False),
Column('name', String),
Column('sex', Integer),
Column('country', String),
)
address = Table('address', metadata,
Column('id', Integer, primary_key=True, autoincrement=False),
Column('user_id', None, ForeignKey('users.id')),
Column('email', String)
)
# 建立
metadata.create_all(engine)
```
如此一般，我們建立起兩個物件，用類似於 SQL CREATE TABLE 的方式創建出兩個table
然後使用 `metadata` 將我們在程式中創建的兩的表的結構對應到真實表的結構，使用 `create_all`會透過我們剛剛建立的連線`engine`連結資料庫，然後檢查這些表是否存在，如果不存在，將會在資料庫上建立一個新的表
* 註 : 從官方文件上找到，假如你的 `primary_key = True`，且你的資料是`Integer`型態的話`autoincrement` 預設會為True，`autoincrement`的功用為當我們在insert 時，這一個`primary_key`的欄位會自動編號下去，如果你的資料表沒有這個功能，要記得關掉
#### Insert Expressions
首先我們先介紹insert要如何使用，程式碼如下
```python=
ins = users.insert()
print(ins)
# INSERT INTO users (id, name, sex, country) VALUES (:id, :name, :sex, :country)
```
非常簡單的就能幫我們創造出一句 SQL 來，但是這裡面還沒有塞入我們要寫入的值進去，下面的程式我們塞一筆值進去看看
```python=
ins = users.insert().values(name='jack', country='Taiwan')
print(ins) # INSERT INTO users (name, country) VALUES (:name, :country)
ins.compile().params # {'name': 'jack', 'country': 'Taiwan'}
```
當我們在values中寫入一筆資料進去後，我們如果要看到我們寫入的值是什麼，我們可以用`ins.compile().params `來獲得我們剛剛寫進去的值
到這裡我們成功創造出一句 insert 的語句了，然而我們還需要讓這句 SQL 執行，才能真正的寫入到我們的資料庫裏面
```python=
conn = engine.connect()
result = conn.execute(ins)
```
首先我們一樣先建立連線，然後再將我們剛剛建立好的`ins`放入`conn.execute()`裡執行即可
我們在資料庫中就會查到一筆我們剛剛寫入的資料
那如何一次插入多筆資料呢?
```python=
conn.execute(address.insert(), [
{'id':1, 'user_id': 1, 'email' : 'jack@yahoo.com'},
{'id':2, 'user_id': 1, 'email' : 'jack@msn.com'}
])
```
如此一來就可以一次寫入多筆資料囉 ~
#### UPDATE、DELETE
接下來我們來學如何 `UPDATE` 和`DELETE`
其實都非常簡單，只需要呼叫 `update()` 和 `delete()`，具體作法如下
```python=
from sqlalchemy import update
u = update(users).where(users.c.id == 5).values(name='jack')
# UPDATE user_r SET name=:name WHERE user_r.id = :id_1
from sqlalchemy import delete
d = delete(users).where(users.c.id==1)
# DELETE FROM user_r WHERE user_r.id = :id_1
```
#### Selecting
在我們學完 `Insert`、`UPDATE` 和`DELETE` 以後，就要來學習如何 `Select` ，`Select` 因為有各種狀況，所以我們先從最簡單的如何 `Select` 出一張表開始吧!
```python=
from sqlalchemy.sql import select
s = select(users)
# 如果是舊版的 sqlalchemy 會報錯
# 改成s = select([users.c.id, users.c.name, users.c.sex, users.c.country]) 即可
result = conn.execute(s)
for row in result:# 讀出
print(row)
# (1, 'jack ', 1, 'Taiwan ')
# 或是這樣讀取
#row = result.fetchone()
#print("name:", row._mapping['name'], "; country:", row._mapping['country'])
# 關閉好習慣
result.close()
```
```python=
# 如果只想輸出特定欄位的話
s = select(users.c.name, users.c.country)
result = conn.execute(s)
for row in result:
print(row)
```
非常的簡單，但是如果我們想要再加上一些條件的話要怎麼做呢?
```python=
s = select(users, address).where(users.c.id == address.c.user_id)
result = conn.execute(s)
for row in result:
print(row)
#(1, 'jack ', 1, 'Taiwan ', 1, 1, 'jack@yahoo.com ')
#(1, 'jack ', 1, 'Taiwan ', 2, 1, 'jack@msn.com ')
```
只要加上 `where` 就可以設下條件，我們也可以看到，當我們將兩個Column做邏輯運算的話，他可以產出如下的語句
```python=
print(users.c.id == address.c.user_id) # user_r.id = address.user_id
# 大於、小於、不等於
print(users.c.id < address.c.user_id) #user_r.id < address.user_id
print(users.c.id > address.c.user_id) #user_r.id > address.user_id
print(users.c.id != address.c.user_id) #user_r.id != address.user_id
# 直接比較資料也是可以的
print(users.c.id != 7) # user_r.id != :id_1
# 等於None 轉換成SQL語句就是 is null
print(users.c.name == None) # user_r.name IS NULL
print('fred' > users.c.name) # user_r.name < :name_1
```
除了邏輯運算以外，也可以做加減乘除
```python=
print(users.c.id + address.c.id # user_r.id + address.id
print(users.c.id - address.c.id # user_r.id - address.id
print(users.c.id * address.c.id # user_r.id * address.id
print(users.c.id / address.c.id # user_r.id / address.id
# String 相加
print(users.c.name + address.c.email) # user_r.name || address.email
```
除此之外，還有like
```python=
print(users.c.name.like('%ed%')) # user_r.name LIKE :name_1
print(users.c.name.ilike('%ed%')) # lower(user_r.name) LIKE lower(:name_1)
# lower 僅能在有提供該函數的SQL server上使用
```
in
```python=
print(users.c.name.in_(['willy','jason'])) # user_r.name IN ([POSTCOMPILE_name_1])
print(~users.c.name.in_(['willy','jason'])) # (user_r.name NOT IN ([POSTCOMPILE_name_1]))
```
is
```python=
print(users.c.name.is_('willy')) # user_r.name IS :name_1
print(users.c.name.is_not('willy')) # user_r.name IS NOT :name_1
```
AND
```python=
from sqlalchemy import and_
print(and_(users.c.name == 'ed', users.c.country == 'USA'))
# user_r.name = :name_1 AND user_r.country = :country_1
```
OR
```python=
from sqlalchemy import or_
print(or_(users.c.name == 'ed', users.c.country == 'USA'))
user_r.name = :name_1 OR user_r.country = :country_1
```
match
```python=
print(users.c.name.match('ed')) # user_r.name MATCH :name_1
```
而除了使用以上幾種他們提供的方式進行 Select 語句的組合撰寫以外，我們其實也可以直接寫一個 SQL 語句，讓`conn.execute()`執行
```python=
from sqlalchemy.sql import text
s = text(
"SELECT users.name || ', ' || address.email AS title "
"FROM users, address "
"WHERE users.id = address.user_id "
"AND (addresses.email_address LIKE :e1 "
"OR addresses.email_address LIKE :e2)")
conn.execute(s, {"x":"m", "y":"z", "e1":"%@yahoo.com%", "e2":"%@msn.com%"}).fetchall()
```
`fetchall()`就是一次全部返回出來，也有`fetchone()`，代表一筆一筆取出資料
再來還有Group by 、 Order by
```python=
from sqlalchemy import func
stmt = select(
address.c.user_id,
func.count(address.c.id).label('num_addresses')).\
group_by("user_id").order_by("user_id", "num_addresses")
conn.execute(stmt).fetchall()
```
`.label()`就是設置別名，`func.count()`等於SQL中`count()`
還有 `func.sum()`等於SQL中`sum()`，`func.max()`等於SQL中`max()`，`func.min()`等於SQL中`min()`
asc()，desc()
```python=
from sqlalchemy import func, desc
stmt = select(
address.c.user_id,
func.count(address.c.id).label('num_addresses')).\
group_by("user_id").order_by("user_id", desc("num_addresses"))
conn.execute(stmt).fetchall()
```
如果我們要幫table取別名，可以這樣做
```python=
a1 = address.alias('a1')
a2 = address.alias('a2')
s = select(users.c.id).\
where(and_(
users.c.id == a1.c.user_id,
users.c.id == a2.c.user_id,
a1.c.email == 'jack@msn.com',
a2.c.email == 'jack@yahoo.com'
))
conn.execute(s).fetchall()
```
轉成SQL語句變成這樣
```sql=
SELECT id
FROM user_r,address as a1, address as a2
WHERE user_r.id=a1.user_id
AND user_r.id=a2.user_id
AND a1.email='jack@msn.com'
and a2.email='jack@yahoo.com'
```
子查詢 subquery()
```python=
sub_usr_id = s.subquery()
s = select(users.c.name).where(users.c.id == sub_usr_id.c.id)
conn.execute(s).fetchall()
```
轉成SQL語句變成這樣
```sql=
SELECT name FROM user_r
where id =(
SELECT id
FROM user_r,address as a1, address as a2
WHERE user_r.id=a1.user_id
AND user_r.id=a2.user_id
AND a1.email='jack@msn.com'
and a2.email='jack@yahoo.com'
)
```
我們學會大部分的 SELECT 語句該怎麼組成後，我們繼續往JOIN前進
當我們需要創造一個JOIN或是OUTERJOIN的時候，我們使用`FromClause.join()`和 `FromClause.outerjoin()`方法
```python=
print(users.join(address))
# user_r JOIN address ON user_r.id = address.user_id
```
`join()`方法可以將你指定的兩個表連接，並且如果你沒有指定要如何連接的話，它會自動尋找是否有指定`ForeignKey`來進行連接，當然如果你想要自己指定連接的方法也是可以
```python=
print(users.join(address,
address.c.email.like(users.c.name + '%')
)
)
# user_r JOIN address ON address.email LIKE user_r.name || :name_1
```
當我們建構好我們的`JOIN`語法後，我們需要用`select_from()`告訴`select()`我們要查詢的表範圍在哪裡
```python=
s = select(users.c.name).select_from(
users.join(address,
address.c.email.like(users.c.name + '%')
)
)
conn.execute(s).fetchall()
```
`FromClause.outerjoin()`也是相同的方法，他輸出的會是`LEFT OUTER JOIN`
```python=
s = select(users.c.name).select_from(
users.outerjoin(address,
address.c.email.like(users.c.name + '%')
)
)
conn.execute(s).fetchall()
```
Common Table Expressions (CTE)
大部分現代的SQL server都有這樣的一種寫法，他看起來會更簡潔，將子查詢制定在最上面，而且我們可以在任何位置引用他，就像引用一個普通的表一般
```python=
users_cte = select(users.c.id, users.c.name).where(users.c.name == 'jack').cte()
stmt = select(address).where(address.c.user_id == users_cte.c.id).order_by(address.c.id)
conn.execute(stmt).fetchall()
```
在轉成 SQL 語句後就是長這樣
```sql=
WITH anon_1 AS
(SELECT user_r.id AS id, user_r.name AS name
FROM user_r
WHERE user_r.name = :name_1)
SELECT address.id, address.user_id, address.email
FROM address, anon_1
WHERE address.user_id = anon_1.id ORDER BY address.id
```
當我們構造一個 CTE 對象時，我們像使用其他 Table 一樣使用它。然而，它不是作為子查詢添加到 FROM 子句中，而是排在最前面，這樣的好處是不會引起意外笛卡爾積。
func
SQL 函數是使用`func`關鍵字創建的，該關鍵字使用屬性訪問生成函數：
```
from sqlalchemy.sql import func
print(func.now())
```
這個方法也就是說我們隨便創造一個函數他也會創造出來給你
```
print(func.qwertsfshcb())
# qwertsfshcb()
```
OVER
`SUM(BBB) OVER(PARTITION BY AAA)`
`OVER` 子句內的`PARTITION BY`可以指定AAA欄位做分割，被分割的會自成一個群組，並將該群組的BBB欄位帶進函式計算。
```python=
s = select(
users.c.id,
func.row_number().over(partition_by=users.c.name,order_by=users.c.name)
)
print(s)
```
轉成 SQL 後
```sql=
SELECT user_r.id, row_number() OVER (PARTITION BY user_r.name ORDER BY user_r.name) AS anon_1
FROM user_r
```
除此之外，還有很多的語法，我們就不繼續介紹下去了
下一章我們來介紹 ORM 的寫法
###### tags: `python`
Sign in with Wallet
Connect another wallet