用Python可以不用記住任何一句mysql，通過Python使用的ORM模型，簡單來說就是一個庫是一個物件，這個庫裡面的欄位是這個物件的屬性。

ORM解釋：物件關係對映（英語：(Object Relational Mapping，簡稱ORM，或O/RM，或O/R mapping），是一種程式技術，用於實現面向物件程式語言裡不同型別系統的資料之間的轉換 。從效果上說，它其實是建立了一個可在程式語言裡使用的–“虛擬物件資料庫”。簡單來說就是：一個庫是一個物件，這個庫裡面的欄位是這個物件的屬性。

所有mysql語句封裝在了這個物件的方法之中，直接使用相關方法。

介紹三款ORM模型：Peewee、Sqlalchemy、mongoengine以及其中遇到過的莫名問題

其中Peewee、Sqlalchemy是操作mysql和sqlite的，mongoengine是操作mongodb的，這三個庫均可在Python3環境下使用

示例簡介使用ORM模型實現資料庫的增刪改查，以及根據ORM模型對映建立資料庫

1.Peewee

from peewee import *
from playhouse.db_url import connect
from playhouse.pool import PooledMySQLDatabase, PooledDatabase

from
 
 playhouse.shortcuts import model_to_dict, dict_to_model
import datetime

"""peewee提供了一個數據庫的封裝模型，playhouse.db_url為其連線的一種方式通過資料庫的指定格式url連線
連線後建立完以後需要模型生成表使用db.connect()，db.create_tables([Person, Pet])"""
blog = PooledMySQLDatabase(
    'peewee_test'
 
,
    max_connections=8,
    stale_timeout=300,
    user='admin',
    host='118.24.52.111',
    password='[email protected]#,.',
    port=3306)



""""
http://docs.peewee-orm.com/en/latest/peewee/playhouse.html#PooledDatabase.manual_close
"""
# MYSQL_URL = \
#     'mysql://admin:[email protected]:3306/peewee_test?charset=utf8'
# blog = connect(MYSQL_URL)   # 連線方式一
# 連線方式二
# blog = MySQLDatabase('test', user='root', host='localhost', port=3306)
class BaseModel(Model):
    """基類"""
    class Meta:
        database = blog


class Test(BaseModel):
    """引數解釋    CharField：字串型別    IntegerField：整型    DateTimeField：時間型別    ForeignKeyField：外來鍵關聯    unique：是否唯一    max_lenth:最大長度    verbose_name:表頭名    null：是都為空    default：預設值"""
    name = CharField(unique=True, max_length=50, verbose_name='使用者名稱', null=False, default='你哈')
    number = IntegerField(default=0, verbose_name='數字')
    update_date = DateTimeField(verbose_name='更新時間', default=datetime.datetime.now)

    def close(self):
        blog.close()


class Tests(BaseModel):
    title = CharField(verbose_name='標題', max_length=64)
    site = CharField(verbose_name='字首', max_length=32, unique=True)
    article_type = ForeignKeyField(Test)


def create_tables():
    """生成資料表，在資料庫中生成模型對映的表"""
    blog.connect()
    blog.create_tables([Tests, Test])
    blog.close()


def drop_tables():
    """刪除資料表"""
    blog.connect()
    blog.drop_tables([Tests, Test])
    blog.close()


def insert(value):
    """插入資料,或者將屬性作為引數傳入Test（name='name',number=2222）"""
    obj = Test()
    obj.name = value
    obj.number = 99
    obj.save()


def updata():
    """"更新資料"""
    obj = Test.get(Test.name == '更新')
    obj.name = '更新完畢'
    obj.number = 100
    obj.save()


def select_all():
    """查詢所有資料"""
    ret = Test.select()
    for obj in ret:
        print(obj.name)


def select_test():
    """查詢條件資料"""
    Test.select().where((Test.name.contains('測試'))) .count()    # 包含指定內容返回集合
    Test.select().where((Test.name == '測試') | (Test.number == 9999)).first()    # 條件或
    Test.select().where((Test.name == '測試'), (Test.number == 9999)).first()    # 條件並
    Test.select().join(Tests).where(Tests.title == 'title').execute()   # 關聯查詢
    obj = Test.get(Test.name == 'yang')
    if obj:
        print(obj.name)
    else:
        print('none have')


def delete_a(i):
    """刪除資料"""
    obj = Test.get(Test.id == i)
    obj.delete_instance()


def sort():
    """對返回結果列排序"""
    set = Test.select().order_by(Test.name)


def to_dict():
    """把模型資料轉為字典物件"""
    user = Test.create(username='jack')
    u = model_to_dict(user)
    return u


def to_model():
    """生成model物件"""
    user_data = {'id': 2, 'username': 'charlie'}
    user = dict_to_model(Test, user_data)


def close():
    db.close()

2.Sqlalchemy

from sqlalchemy import Column, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.types import CHAR, Integer, String, SMALLINT

"""
__tablename__：指定表名Column：行宣告，可指定主鍵Integer：int型別String：字串Float：浮點Boolean：布林DateTime：日期和時間Text：文字LongText：長文字
父類BaseModel會呼叫所有繼承他的子類來生成表結構primary_key：True/False 是否為主鍵unique：是否唯一nullable：是否為空default：預設值"""
MYSQL_SETTING = "mysql://lkj:[email protected]:3306/blog?charset=utf8"
engine = create_engine(
    MYSQL_SETTING, pool_size=20, max_overflow=0, pool_recycle=3600)

DBSession = sessionmaker(bind=engine)   # 建立DBSession型別:類似資料庫連線
BaseModel = declarative_base()


class Test(BaseModel):

    __tablename__ = 'test_table'
    id = Column(Integer, primary_key=True)
    url = Column(String(128))   # 字串型別
    update_time = Column(DateTime, default=datetime.datetime.now())

    def to_dict(self):  # 將讀取的資料和轉化成字典
        return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}


def init_db():
    """生成資料表"""
    BaseModel.metadata.create_all(engine)


def drop_db():
    """刪除資料表"""
    BaseModel.metadata.drop_all(engine)


if __name__ == '__main__':
    init_db()

    """新增資料"""
    session = DBSession()
    new_data = Test(url='', update_time='')
    session.add(new_data)  # 新增
    session.commit()  # 提交
    """查詢資料"""
    session = DBSession()
    num = session.query(Test).filter(
        Test.url == 'url', Test.id == '22').count()    # 查詢多條件
    session.commit()
    session.close()

    """刪除資料"""
    session = DBSession()
    test = session.query(Test).filter(Test.url == "user1").first()
    session.delete(test)
    session.commit()
    session.close()

    """更新資料"""
    session = DBSession()
    test = session.query(Test).filter(Test.url == "user1").first()
    test.url = 'www.test.com'
    session.commit()
    session.close()

3.mongoengine

from mongoengine import connect, Document, EmbeddedDocument, DynamicDocument, \
    StringField, IntField, FloatField, ListField, EmbeddedDocumentField, DictField
import datetime

connect(db='test',
        host="mongodb://admin:[email protected]:27017/?authSource=admin")


if __name__ == '__main__':

    """
    簡單使用說明-及其使用案例    
    Document    #定義基本模式繼承該類——適用於儲存字典結構變化不定的資料    EmbeddedDocument    #申明內嵌文件    EmbeddedDocumentField   #嵌入文件的方法    DynamicDocument # 動態新增欄位方法--適用於儲存固定格式字典並要求驗證的    

    StringFiled(regex=None,max_length=None,min_lenght=None) #字串型別 
    IntField(min_value=None,max_value=None) #整數型別 
    FloatField(min_value=None,max_value=None) #字串型別 
    BooleanField() #布林型別 
    DateTimeField() #時間型別 
    listField() #可以插入列表的 
    DictField() #字典型別 
    ReferenceField() #參照型別 
    SequenceField() #自動產生一個數列、 遞增的

    通用引數    default #預設值 也可以是一個函式 可呼叫型別 
    required #是否必須賦值 true false 
    primary_key #插入資料是否重複 
    null #賦值是否可以為空 
    choices #列表的範圍 
    unique #當前列只能是唯一的    """
    SEX_CHICES = (
        ('male', '男'),
        ('female', '女')
    )


    class Grade(EmbeddedDocument):
        """成績"""
        name = StringField(required=True)
        score = FloatField(required=True)


    class Student(DynamicDocument):
        """學生"""
        name = StringField(max_length=32, required=True)
        age = IntField(required=True)
        sex = StringField(choices=SEX_CHICES, required=True)
        grade = FloatField()
        address = StringField()
        grades = ListField(EmbeddedDocumentField(Grade))

        meta = {
            'collection': 'students',
            # 排序功能，按照分數倒序
            'ordering': ['-grade']
        }


    class TestMongoEngine(object):

        def add_one(self):
            """新增一條資料到資料庫"""
            yuwen = Grade(
                name='語文',
                score=90)
            shuxue = Grade(
                name='數學',
                score=100)
            stu_obj = Student(
                name='張三丰',
                age=15,
                grades=[yuwen, shuxue],
                sex='male'
            )
            # 直接新增remark欄位是無法新增成功的，需要引入動態新增欄位的方法DynamicDocument
            stu_obj.remark = 'remark'
            stu_obj.save()
            return stu_obj

        def get_one(self):
            """ 獲取單條資料 """
            return Student.objects.first()

        def get_more(self):
            """ 獲取多條資料 """
            # return Student.objects
            return Student.objects.all()

        def get_one_from_oid(self, oid):
            """ 查詢指定id的資料 """
            return Student.objects.filter(id=oid).first()

        def update(self):
            """ 修改資料 """
            # 修改一條資料
            # res = Student.objects.filter(sex='male').update_one(inc__age=1)
            # return res
            # 修改多條資料
            res = Student.objects.filter(sex='male').update(inc__age=10)
            return res

        def delete(self):
            """ 刪除資料 """
            # 刪除一條資料
            # res = Student.objects.filter(sex='male').first().delete()
            # return res
            # 刪除多條資料
            res = Student.objects.filter(gender='male').delete()

    test = TestMongoEngine()
    test.add_one()

注意：peewee和sqlalchemy如果使用的是單個連線在多程序使用中會鎖表，上面案列連線方式都是連線池的方式，可以避免在多程序中鎖表。

ID：Python之戰

|作|者|公(zhong)號:python之戰

專注Python，專注於網路爬蟲、RPA的學習-踐行-總結

喜歡研究和分享技術瓶頸，歡迎關注

獨學而無友,則孤陋而寡聞！