粗糙版ORM（附詳細注釋）
- 2019 年 10 月 7 日
- 筆記
目錄
ORM
作為數據庫表記錄 和 python中對象的映射關係中間件
|
數據庫中 |
python代碼中 |
|---|---|
|
不同的表 |
不同的表模型類 |
|
一條條記錄 |
一個個模型類對象 |
|
記錄里的某個字段 |
模型類對象的屬性 |
在python代碼中通過操作orm來進行數據庫的存取操作
這為簡易版demo，查詢條件等不夠完善，僅展示實現原理
其他
焦急規劃中…
ORM代碼
數據庫表代碼
數據庫使用 mysql，將下面的 mysql代碼導入數據庫 需先 安裝 pymysql 模塊 在
db/pymysql_opreator.py中把 pymysql 配置那塊兒更改數據庫，密碼等
mysql代碼
/* Navicat MySQL Data Transfer Source Server : localhost-E Source Server Type : MySQL Source Server Version : 50645 Source Host : localhost:3306 Source Schema : youkuserver Target Server Type : MySQL Target Server Version : 50645 File Encoding : 65001 Date: 27/08/2019 08:35:35 */ SET NAMES utf8mb4; SET FOREIGN_KEY_CHECKS = 0; -- ---------------------------- -- Table structure for user -- ---------------------------- DROP TABLE IF EXISTS `user`; CREATE TABLE `user` ( `id` int(11) NOT NULL AUTO_INCREMENT, `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, `password` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL, PRIMARY KEY (`id`) USING BTREE ) ENGINE = InnoDB AUTO_INCREMENT = 15 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact; -- ---------------------------- -- Records of user -- ---------------------------- INSERT INTO `user` VALUES (1, 'tank_dsb', '123456'); INSERT INTO `user` VALUES (2, 'jason', '123'); SET FOREIGN_KEY_CHECKS = 1;
db/models.py
from db.pymysql_opreator import MyMySQL # pycharm默認將 項目根目錄添加到 sys.path中，這裡就先不寫啟動文件和配置文件的了，後續完善也會要寫的 ''' 定義幾個字段類 int --》 IntegerField varchar --》 StringField ''' class Field: def __init__(self, field_name, field_type, is_primary_key, default_value): self.field_name = field_name self.field_type = field_type self.is_primary_key = is_primary_key self.default_value = default_value # 先定義幾個常見的字段類型 class IntegerField(Field): # 這後面的類型要是數據庫的類型，用來拼接 sql語句的， 可能有多個 int 類型，那 is_primary_key=False 默認值得是 False， default 該字段的默認值， 0 def __init__(self, field_name, field_type='int', is_primary_key=False, default_value=0): super().__init__(field_name, field_type, is_primary_key, default_value) class StringField(Field): # 這後面的類型要是數據庫的類型，用來拼接 sql語句的， 可能有多個 int 類型，那 is_primary_key=False 默認值得是 False， default 該字段的默認值， 0 def __init__(self, field_name, field_type='varchar(255)', is_primary_key=False, default_value='NULL'): super().__init__(field_name, field_type, is_primary_key, default_value) ''' 表類 有且只有一個主鍵字段 表名必須有 所有的字段專門放在 mappings 裏面 特殊點 不確定字段個數 ---> 規定採用關鍵字傳參的方式來寫，利用字典可以接收任意個數關鍵字參數的特性 要支持點語法 ---> 對象.屬性 觸發的是 __getattr__、 __setattr__、 __delattr__ 這幾個方法，重寫一下，把字典的值返回回去，實現點語法的支持 意向語法 生成一張表 --> 直接定義一個表類 一條記錄 --> 表類實例化 --> user_obj = User(name='tank', password='123') 一個字段 --> 類對象.屬性 --> user_obj.name user_obj = User(name='tank', password='123') user_obj.insert_record() --> 新增一條記錄 user_obj.update_record() --> 更新一條記錄 user_obj.select_record() --> 查詢記錄 列表套對象 user_obj.select(name='tank') --> 找出 name='tank' 的記錄 列表套對象 user_obj.name --> 獲取到 name屬性的值 user_obj.name = 'tank_dsb' --> 更新屬性值 user_obj.age = 18 --> 報錯！ 表結構已固定，不需要額外的 user_obj.table_name --> 獲取表名 user_obj.primary_key_field --> 主鍵字段名 user_obj.name.is_primary_key --> name 字段是否是主鍵 user_obj.name.field_name --> 字段的字段名 user_obj.name --> 獲取到 name屬性的值 ''' class MyMetaClass(type): # ---------------------------------------------------------- # 表類限制 # 有且只有一個主鍵字段 # 表名必須有 ---> 不一定表模型類的類名就能對應上數據庫中表名，建議還是匹配上 # 所有的字段專門放在 mappings 裏面 # # 控制類的創建過程 # ---------------------------------------------------------- # def __new__(cls, *args, **kwargs): # 默認 **kwargs 是空的，而 args 得到的是 類名(字符串)、類的父類們(元組)、名稱空間(字典) def __new__(cls, class_name, class_bases, class_attr: dict): # 把接收到的 args 解壓賦值給 class_name, class_bases, class_attr # ----------------------------------------------------------- # 過濾表模型類 # ----------------------------------------------------------- if class_name == 'Models': return super().__new__(cls, class_name, class_bases, class_attr) # MyMetaClass 的父類就是元類 # return type.__new__(cls, class_name, class_bases, class_attr) # 寫法二 # ----------------------------------------------------------- # 表名必須有 # ----------------------------------------------------------- try: table_name = class_attr['table_name'] except Exception as e: print(e) raise Exception(f'表模型類 {class_name} 中沒有指定 表名 table_name') # table_name = class_attr.get('table_name', f'表模型類 {class_name} 中沒有指定 表名 table_name') # ----------------------------------------------------------- # 有且只有一個主鍵字段、所有的字段專門放在 mappings 裏面 # ----------------------------------------------------------- primary_key_field = None mappings = {} # 將表字段都放進來 for key, value in class_attr.items(): # 循環遍歷 名稱空間(字典)中的鍵值對 # 只處理字段類型的鍵值對 if isinstance(value, Field): # 是不是主鍵 if value.is_primary_key: # 只能有一個主鍵過濾 if primary_key_field: raise Exception(f'一個表只能有一個主鍵！ 出錯表模型：{class_name}') primary_key_field = value.field_name mappings[key] = value # {'name': 'tank', 'password': '123'} if not primary_key_field: raise Exception(f'每個模型表都必須有一個主鍵！ 出錯表模型：{class_name}') # 把名稱空間中的字段鍵值對刪除掉（節省點空間，字段鍵值對都放在mappings 裏面了） for key, value in mappings.items(): class_attr.pop(key) # 本來就是名稱空間里取過來的字段鍵值對，所以這裡不會報錯 # 將 primary_key_field、mappings 放到名稱空間里，接着實例化成一個類 # table_name 本身就在 class_attr 里， 這一塊只是限制每個表模型類必須指定表名 class_attr['primary_key_field'] = primary_key_field class_attr['mappings'] = mappings # 會調用 Models的 __new__方法，最後走到MyMetaClass 的 __new__， 而 class_name 是models，則會直接調用 type的 __new__方法 return super().__new__(cls, class_name, class_bases, class_attr) # return type.__new__(cls, class_name, class_bases, class_attr) # 寫法二 class Models(dict, metaclass=MyMetaClass): # ---------------------------------------------------------- # 利用字典可以接收任意個數關鍵字參數的特性來保存字段 # user_obj = User(name='tank', password='123') # ---------------------------------------------------------- def __init__(self, **kwargs): # 將接收到的關鍵字參數 通過 **kwargs 打包成字典 --> kwargs # 打散傳給 dict 轉為字典屬性 --> dict(name='tank', password='123') --> {name='tank', password='123'} super().__init__(self, **kwargs) # ---------------------------------------------------------- # 支持點語法 對象.屬性 # ---------------------------------------------------------- # user_obj.name --> 獲取到 name屬性的值 def __getattr__(self, item): return self.get(item, f'表模型類 {self.__class__.__name__} 中沒有此字段 {item}') # try: # return self[item] # except Exception as e: # print(e) # raise Exception(f'表模型類 {self.__class__.__name__} 中沒有此字段 {item}') # user_obj.name = 'tank_dsb' --> 更新屬性值 # user_obj.age = 18 --> 報錯！ 表結構已固定，不需要額外的 ---> 先不管 def __setattr__(self, key, value): self[key] = value # ---------------------------------------------------------- # 支持點查詢、更改、插入方法 # ---------------------------------------------------------- # user_obj.select_record() --> 查詢記錄 列表套對象 # user_obj.select(name='tank') --> 找出 name='tank' 的記錄 列表套對象 @classmethod # 一般不會拿着對象（記錄）去查記錄（對象），所以這裡寫成類綁定方法 def select_record(cls, **kwargs): # ---> 這裡暫時只做一個查詢條件， 且是 ... = ... 的形式 # 只在調用的時候才打開這個 mysql_op = MyMySQL() # select * from user; # 或者 select * from user where name='tank' 多字段暫時不考慮 sql = f'select * from {cls.table_name}' if not kwargs: # 如果沒有條件 res = mysql_op.select_record(sql) else: # select * from user where name='tank' # {name='tank', id=1} field = list(kwargs.keys())[0] # kwargs.keys() --> dict_keys(['name', 'age']) 需要list 強轉成列表 value = list(kwargs.values())[0] sql = f'{sql} where {field}=%s' res = mysql_op.select_record(sql, value) # res = [{}, {}] 返回的都是列表套字典，現在要轉換成 列表套對象 record_objs = [] for record_dic in res: # cls(**record_dic) # {'name': 'jason', 'password': '123'} --> name='jason' password='123' # 將字典打散，拆分成一個個關鍵字參數的形式 record_objs.append(cls(**record_dic)) return record_objs # user_obj.insert_record() --> 新增一條記錄 def insert_record(self): # insert into user (name, password) values('tank', '123') id 自動填入 mysql_op = MyMySQL() fields = [] values = [] for key, value in self.mappings.items(): # 主鍵默認是id 默認自動增長，所以這裡忽略掉 if not value.is_primary_key: fields.append(key) values.append(getattr(self, value.field_name, value.default_value)) # ['egon','123'] # 這裡單雙引號千萬不要包錯了， (".".join(fields) --> (name, password) list_s = ['%s' for i in range(len(fields))] # insert into user (name,password) values (%s,%s) # ({",".join(list_s)}) --> (%s, %s) sql = f'insert into {self.table_name}({",".join(fields)}) values ({",".join(list_s)})' mysql_op.execute_sql(sql, values) # user_obj.update_record() --> 更新一條記錄 def update_record(self): # 不能取名 update ---> 和字典的 update 方法重名，會造成方法重寫，覆蓋 # update table user set name='tank_dsb', passowrd='123' where id=1 mysql_op = MyMySQL() primary_key_field = self.primary_key_field primary_key_value = 0 fields = [] values = [] for key, value in self.mappings.items(): # 主鍵默認是id 默認自動增長，所以這裡忽略掉 if not value.is_primary_key: fields.append(key) # values.append(value) # 報錯, ---> execute 拼接的是字符串，拼接不了對象(這裡 value 是 StringField） values.append(getattr(self, value.field_name, value.default_value)) # 主鍵存起來做判斷條件 else: primary_key_value = getattr(self, self.primary_key_field, value.default_value) # primary_key_value = getattr(self, value.field_name, value.default_value) # 這裡單雙引號千萬不要包錯了， (".".join(fields) --> (name, password) fields_s = [f"{field}=%s" for field in fields] # ",".join(fields_s) --> name=%s, passowrd=%s sql = f'update {self.table_name} set {",".join(fields_s)} where {primary_key_field}={primary_key_value}' # print(sql, '------------------------------') # # update user set name=%s,password=%s where id=15 ------------------------------ # update user set name='tank_dsb', passowrd='123' where id=1 mysql_op.execute_sql(sql, values) if __name__ == '__main__': class User(Models): table_name = 'user' id = IntegerField('id', is_primary_key=True) name = StringField('name') password = StringField('password') # print(User.select_record(name='tank')) # 打印按條件查詢 # select * from user where name='tank' # [{'id': 1, 'name': 'tank', 'password': '123'}] print(User.select_record()) user_obj = User(name='egon', password='123') user_obj.insert_record() user_obj = User.select_record(name='egon')[0] # 重名的時候會有問題 （兩個 egon，改的是最前面那條 egon的記錄），我也暫時只做了一個 查詢條件 # print(user_obj.name, '++++++++++') # # egon ++++++++++ user_obj.password = '666' user_obj.update_record() user_egon = User.select_record(name='egon')[0] # 取出來的是 列表 ，我們只要第一個元素（就算只有一條記錄也是一個列表） # select * from user # [{'id': 1, 'name': 'tank', 'password': '123'}, {'id': 2, 'name': 'jason', 'password': '123'}]
db/pymysql_opreator.py
import pymysql ''' 數據庫連接類 包含數據庫連接和關閉數據庫方法 ''' class MyMySQL: # ------------------------------------------------- # 單例模式 # ------------------------------------------------- _instance = None # __init__ 不能返回任何東西，所以寫在這裡，當然其他地方也行 def __new__(cls, *args, **kwargs): if not cls._instance: cls._instance = object.__new__(cls, *args, **kwargs) return cls._instance # ------------------------------------------------- # 初始化建立數據庫連接，獲得游標對象 # ------------------------------------------------- def __init__(self): # 建立mysql連接 self.conn = pymysql.connect( host='127.0.0.1', port=3306, user='root', password='000000', database='YouKuServer', charset='utf8', autocommit=True ) # 拿到游標對象 self.cursor = self.conn.cursor(pymysql.cursors.DictCursor) # 關閉資源操作 def close(self): # 一定要先關閉游標再關閉連接 self.cursor.close() self.conn.close() # def __del__(self): # 噹噹前類對象被刪除時自動觸發，關閉資源（可能多線程多進程會對其他的造成影響吧？） # self.close() def select_record(self, sql, args=None): try: print(self.cursor.mogrify(sql, args)) # 打印執行的sql語句 # select * from user where name='egon' self.cursor.execute(sql, args) return self.cursor.fetchall() except Exception as e: print(e) raise Exception(f'這裡報錯啦！{self.__class__.__name__}') # 這裡為什麼要分開來？ 就因為有沒有返回值嗎？ def execute_sql(self, sql, args): try: print(self.cursor.mogrify(sql, args)) # 打印執行的sql語句 # # update user set name='egon',password='666' where id=15 self.cursor.execute(sql, args) except Exception as e: print(e) raise Exception(f'這裡報錯啦！{self.__class__.__name__}')
