mongdb

from pymongo import MongoClient
from bson import ObjectId
# _id是mongodb自動生成的id，其型別為ObjectId，想要使用就需要轉換型別。

#help 遠端的話 需要這樣操作
#uri = "mongodb://%s:%[email protected]%s" % (
#quote_plus(user), quote_plus(password), quote_plus(socket_path))
#client = MongoClient(uri)

conn = MongoClient()  #本地可以不填
db = conn.test  #連線測試資料庫，沒有則自動建立

##繫結 一個集合
def get_collection(collection_name):
    return db.get_collection(collection_name)

##根據ID查詢
def get_one_by_id(collection_name, id):
    collect_cur = get_collection(collection_name)
    return collect_cur.find_one({'_id': ObjectId(id)})

#查詢一條資訊
def get_one(collection_name, ifmation):
    collect_cur = get_collection(collection_name)
    return collect_cur.find_one(ifmation)

##查詢所有資訊
def get_all(collection_name, ifmation=None):
    collect_cur = get_collection(collection_name)
    return collect_cur.find(ifmation)
# 插入多條記錄 ##這裡的data 為陣列
def insert_ifmation(collection_name, data):
    collect_cur = get_collection(collection_name)
    return collect_cur.insert_many(data)

# 更新一條資訊
def update_one(collection_name, ifmation, data):
    collect_cur = get_collection(collection_name)
    return collect_cur.update_one(ifmation, {"$set": data})

# 更新所有資訊
def update_many(collection_name, filter, data):
    collect_cur = get_collection(collection_name)
    return collect_cur.update_many(filter, {"$set": data})

##刪除資訊
def delete(collection_name, ifmation):
    collect_cur = get_collection(collection_name)
    return collect_cur.remove(ifmation)

##刪除所有
def del_all(collection_name):
    collect_del_all=get_collection(collection_name)
    collect_del_all.remove()

if __name__ == '__main__':
    # collection_name=get_collection(conn)

    conn = MongoClient()
    # 第二步 獲取指定的資料庫例項
    db = conn.test  # 連線study資料庫，沒有則自動建立
    # 第三步 獲取指定的集合
    study = db.get_collection('text')


 

Mysql

import pymysql
建立遊標  pymysql.connect(配置資訊).cursor()
遊標的常用方法:

 1. close()  關閉
 2. execute()  執行一條SQL語句
 3. executemany()  執行多條SQL語句  傳入一個列表
 4. fetchall()     獲取前面  前面必須使用execute() 進行查詢 返回全部的資訊
 5. fetchmany(3)  可指定返回的條數
 

conn = pymysql.connect(
    host='localhost',
    port=33060,
    user='root',
    password='root',
    db='test', #資料庫名
    charset='utf8'
)


class MySQL():
    '''
    MySQL資料庫的操作
    '''

    ##傳入連線的資料庫
    def __init__(self, conn, table):

        self.conn = conn
        ##建立遊標
        self.cou = self.conn.cursor()
        ##得到表名，資料
        self.table = table

    ##執行你的sql語句
    def finish_you(self, sql=None, *args, **kwargs):
        if sql:
            try:
                self.cou.execute(sql)
                self.conn.commit()
            except Exception:
                print('報錯了')
                self.conn.rollback()
            else:
                print('執行成功')
        else:
            pass

    ##檢視,

    def find_ifmations(self, number=None):
        '''
        :param number: 請輸入檢視資訊的條數，預設檢視全部
        :return:
        '''
        ##檢視前必執行
        sql = 'select * from {} '.format(self.table)
        self.cou.execute(sql)

        if not number:
            li_messages = self.cou.fetchall()

            for li_message in li_messages:
                print('id:%d name:%s age:%d' % li_message)
        else:
            li1_messages = self.cou.fetchmany(number)
            for li_message in li1_messages:
                print('id:%d name:%s age:%d' % li_message)
        print('檢視完畢')

    ##獲取表中一共有多少條資料
    def get_numbers(self):
        number_sql = 'select * from {}'.format(self.table)
        numbers = self.cou.execute(number_sql)
        print('{}這張表,一共有{}條資料'.format(self.table, numbers))

    ##增加多條
    def insert_ifmations(self, contents):

        ##這裡 是 字串，
        inserts_sql = "insert {} values (%s,%s,%s);".format(self.table)
        try:
            self.cou.executemany(inserts_sql, contents)
            self.conn.commit()
        except Exception:
            print('報錯了')
            self.conn.rollback()
        else:
            print('成功插入資料')

    ##增加單條
    def insert_ifmation(self, content):
        inserts_sql = "insert {} values{};".format(self.table, content)
        try:
            self.cou.execute(inserts_sql)
            self.conn.commit()
        except Exception:
            print('報錯了')
            self.conn.rollback()
        else:
            print('成功插入資料')

    ##刪除資訊
    def del_ifmations(self, id):
        del_sql = "delete from {} where id={};".format(self.table, id)
        try:
            self.cou.execute(del_sql)
            self.conn.commit()
        except Exception:
            print('報錯了')
            self.conn.rollback()
        else:
            print('刪除成功')

    ##更新資訊
    def update_ifmations(self, your_message, id):
        update_sql = "update %s set name='%s' where id=%d" % (self.table, your_message, id)
        try:
            self.cou.execute(update_sql)
            self.conn.commit()
        except Exception:
            print('報錯了')
            self.conn.rollback()
        else:
            print('更新成功')

    ##先關閉遊標，再關閉資料庫連線
    def close_mysql(self):
        self.cou.close()
        self.conn.close()
        print('關閉成功')


if __name__ == '__main__':
    table = 'student'

    content_one = "(12,'小小鳥',23)"
    content_many = [(str(i), '大大鳥', '23') for i in range(45, 50)]

    test = MySQL(conn, table, )

    ##測試
    test.insert_ifmations(content_many)
    test.close_mysql()