Python建立SQL資料庫流程逐步講解

2022-09-24 14:01:08

前言

根據《2021年Stackoverflow開發者調查》，

SQL是最常用的五種程式語言之一。

所以，我們應該多投入時間來學習SQL。

由Storyset繪製的人物插圖

但是有一個問題：

如何在沒有資料庫的情況下練習資料庫查詢呢？

在今天的文章中，讓我們一起來解決這個基本問題，學習如何從零開始建立自己的MySQL資料庫。在Python和一些外部庫的幫助下，我們將建立一個簡單的指令碼，可以自動建立並使用隨機生成的資料，填充我們的表格。

但是，在討論實現細節之前，我們首先需要討論一些先決條件。

注意：當然還有其他方法可以獲取用於實踐的SQL資料庫(例如直接找資源下載)，但使用Python和一些外部庫可以為我們提供額外且有價值的實踐機會。

先決條件

我們先從最基本的開始。

首先，需要安裝MySQL Workbench並連線服務，接下來就可以開始建立資料庫：

CREATE DATABASE IF NOT EXISTS your_database_name;

現在，我們只需要安裝必要的python庫，基本的設定就完成了。我們將要使用的庫如下所示，可以通過終端輕鬆安裝。

NumPy： pip install numpy
Sqlalchemy： pip install sqlalchemy
Faker： pip install faker

建立指令碼

完成基本設定後，我們可以開始編寫python指令碼了。

先用一些樣板程式碼建立一個類，為我們提供一個藍圖，指導我們完成其餘的實現。

import numpy as np
import sqlalchemy
from faker import Faker [python學習裙：90 3971231###
from sqlalchemy import Table, Column, Integer, String, MetaData, Date,
class SQLData:
    def __init__(self, server:str, db:str, uid:str, pwd:str) -> None:
        self.__fake = Faker()
        self.__server = server
        self.__db = db
        self.__uid = uid
        self.__pwd = pwd
        self.__tables = dict()
    def connect(self) -> None:
        pass
    def drop_all_tables(self) -> None:
        pass
    def create_tables(self) -> None:
        pass
    def populate_tables(self) -> None:
        pass

目前我們還沒用特別高階的語法。

我們基本上只是建立了一個類，儲存了資料庫憑據供以後使用，匯入了庫，並定義了一些方法。

建立連線

我們要完成的第一件事是建立一個資料庫連線。

幸運的是，我們可以利用python庫sqlalchemy來完成大部分工作。

class SQLData:
    #...
    def connect(self) -> None:
        self.__engine = sqlalchemy.create_engine(
            f"mysql+pymysql://{self.__uid}:{self.__pwd}@{self.__server}/{self.__db}"
        )
        self.__conn = self.__engine.connect()
        self.__meta = MetaData(bind=self.__engine)

這個方法可以建立並儲存3個物件作為範例屬性。

首先，我們建立一個連線，作為sqlalchemy應用程式的起點，描述如何與特定型別的資料庫/ DBAPI組合進行對話。

在我們的例子中，我們指定一個MySQL資料庫並傳入我們的憑據。

接下來，建立一個連線，它可以讓我們執行SQL語句和一個後設資料物件(一個容器)，將資料庫的不同功能放在一起，讓我們關聯和存取資料庫表。

建立表格

現在，我們需要建立資料庫表。

class SQLData:
    #...
    def create_tables(self) -> None:
        self.__tables['jobs'] = Table (
            'jobs', self.__meta,
            Column('job_id', Integer, primary_key=True, autoincrement=True, nullable=False),
            Column('description', String(255))
        )
        self.__tables['companies'] = Table(
            'companies', self.__meta,
            Column('company_id', Integer, primary_key=True, autoincrement=True, nullable=False),
            Column('name', String(255), nullable=False),
            Column('phrase', String(255)),
            Column('address', String(255)),
            Column('country', String(255)),
            Column('est_date', Date)
        )
        self.__tables['persons'] = Table(
            'persons', self.__meta,
            Column('person_id', Integer, primary_key=True, autoincrement=True, nullable=False),
            Column('job_id', Integer, ForeignKey('jobs.job_id'), nullable=False),
            Column('company_id', Integer, ForeignKey('companies.company_id'), nullable=False),
            Column('last_name', String(255), nullable=False),
            Column('first_name', String(255)),
            Column('date_of_birth', Date),
            Column('address', String(255)),
            Column('country', String(255)),
            Column('zipcode', String(10)),
            Column('salary', Integer)
        )
        self.__meta.create_all()

我們建立了3個表，並將它們儲存在一個字典中，以供以後參考。

在sqlalchemy中建立表也非常簡單。我們只需範例化一個新的表，提供表名、後設資料物件，並指定不同的列。

在本例中，我們建立了一個job表、一個company表和一個person表。person表還通過了foreign kkey連結了其他表，這使資料庫在實踐SQL連線方面更加有趣。

定義了所有表格之後，我們只需呼叫MetaData物件的create_all()方法就好了。

生成一些亂資料

雖然我們建立了資料庫表，但仍然沒有任何資料可用。因此，我們需要生成一些亂資料並將其插入到表中。

class SQLData:
    #...
    def populate_tables(self) -> None:
        jobs_ins = list()
        companies_ins = list()
        persons_ins = list()
        for _ in range(100):
            record = dict()
            record['description'] = self.__fake.job()
            jobs_ins.append(record)
        for _ in range(100):
            record = dict()
            record['name'] = self.__fake.company()
            record['phrase'] = self.__fake.catch_phrase()
            record['address'] = self.__fake.street_address()
            record['country'] = self.__fake.country()
            record['est_date'] = self.__fake.date_of_birth()
            companies_ins.append(record)
        for _ in range(500):
            record = dict()
            record['job_id'] = np.random.randint(1, 100)
            record['company_id'] = np.random.randint(1, 100)
            record['last_name'] = self.__fake.last_name()
            record['first_name'] = self.__fake.first_name()
            record['date_of_birth'] = self.__fake.date_of_birth()
            record['address'] = self.__fake.street_address()
            record['country'] = self.__fake.country()
            record['zipcode'] = self.__fake.zipcode()
            record['salary'] = np.random.randint(60000, 150000)
            persons_ins.append(record)
        self.__conn.execute(self.__tables['jobs'].insert(), jobs_ins)
        self.__conn.execute(self.__tables['companies'].insert(), companies_ins)
        self.__conn.execute(self.__tables['persons'].insert(), persons_ins)

現在，我們可以利用Faker庫來生成亂資料。

我們只需在for迴圈中使用隨機生成的資料，建立一個由字典表示的新記錄。然後將單個記錄追加到可用於(多個)insert語句的列表中。

接下來，從連線物件中呼叫execute()方法，並將字典列表作為引數傳遞。

就是這樣！我們成功實現了類—只需要把類範例化，並呼叫相關函數來建立資料庫。

if __name__ == '__main__':
    sql = SQLData('localhost','yourdatabase','root','yourpassword')
    sql.connect()
    sql.create_tables()
    sql.populate_tables()

試著做一個查詢

剩下的唯一一件事是——需要驗證我們的資料庫是否已經啟動和執行，是否確實包含一些資料。

從基本的查詢開始：

SELECT *
FROM jobs
LIMIT 10;

基本查詢結果[圖片by作者]

看起來我們的指令碼成功了，我們有一個包含實際資料的資料庫。

現在，嘗試一個更復雜的SQL語句：

SELECT
  p.first_name,
  p.last_name,
  p.salary,
  j.description
FROM
  persons AS p
JOIN
  jobs AS j ON
  p.job_id = j.job_id
WHERE
  p.salary > 130000
ORDER BY
  p.salary DESC;