简介：

持久化存储

在任何应用中，都需要持久化存储。一般有三种基础的存储机制：文件、数据库系统以及一些混合类型。这种混合类型包括现有系统上的API、ORM(对象关系映射)、文件管理器、电子表格、配置文件等。

数据库基本操作和SQL

底层存储

数据库通常使用文件系统作为基本的持久化存储，它可以是普通的操作系统文件、专用的操作系统文件，甚至是原始的磁盘分区。

用户接口

大多数数据库系统提供命令行工具与GUI工具，用其执行SQL语句或者查询。

数据库

一个关系数据库管理系统（RDBMS）通常可以管理多个数据库，比如销售、市场、用户支持等，都可以在同一个服务端，如果RDBMS基于服务器，可以这样（例如 Mysql），不过一些简单的系统通常不是基于服务器的（例如 SQLite和Gadfly）。

组件

数据库存储可以抽象为一张表。每行数据都有一些字段对应于数据库的列。每一列的表定义的集合以及每个表的数据类型放到一起定义了数据库的模式（schema）。数据库可以创建(create)、删除(drop)，数据库中的表数据可以插入(insert)、更新(update)、删除行(delete)、查询(query)。一些数据库使用游标的概念来提交SQL命令、查询以及获取结果，不管是一次性获取还是逐行获取都可以使用该概念。

SQL

数据库命令和查询操作是通过SQL语句提交到数据库的。并非所有数据库都使用SQL语句，但是大多数关系数据库使用。注意：大部分数据库都是不区分大小写，尤其是针对数据库命令而言。

创建数据库

CREATE DATABASE test；
CRANT ALL ON test.* to user(s);   #可以为指定用户（或所有用户）提升权限

使用数据库

USE test;

删除数据库

DROP DATABASE test;

创建表

CREATE TABLE users (login VARCHAR(8), userid INT, projid INT);

删除表

DROP TABLE users;

插入行

INSERT INTO users VALUES('leanna',2111,1);

更新行

UPDATE users SET projid=4 WHERE projid=2;
UPDATE users SET projid=1 WHERE projid=311;

删除行

DELETE FROM users WHERE projid=%d;
DELETE FROM users;

Python的DB-API

DB-API 是阐明一系列所需对象和数据库访问机制的标准

https://www.python.org/dev/peps/pep-0249/

数据库适配器示例

mysql适配器的文档：https://wiki.python.org/moin/MySQL

python3下安装mysqldb

sudo apt-get install libmysqlclient-dev
sudo pip3 install -U mysqlclient

python3下安装mysql.connector

sudo pip3 install -U mysql-connector

MySQL、SQLite、Gadfly（只测试 MySQL）(ushuffle_db.py)

#!/usr/bin/python3

from distutils.log import warn as printf #distutils.log.warn 替代py2中的print和py3中的print
import os
import builtins
from random import randrange as rand

# isinstance:Return whether an object is an instance of a class or of a subclass thereof.
# builtins该模块提供对Python的所有“内置”标识符的直接访问（判断raw_input是否在dict模块中存在）
if isinstance(__builtins__, dict) and 'raw_input' in __builtins__:
    scanf = raw_input
elif hasattr(__builtins__, 'raw_input'):
    scanf = raw_input
else:
    scanf = input

COLSIZ = 10
FIELDS = ('login', 'userid', 'projid')
RDBMSs = {'s': 'sqlite', 'm': 'mysql', 'g': 'gadfly'}
DBNAME = 'test'
DBUSER = 'root'
#表示数据库异常（DataBase_EXCeption）
DB_EXC = None
NAMELEN = 16

#标题样式格式化函数(左对齐)
def tformat(s): return str(s).title().ljust(COLSIZ)

#全大写格式化函数(左对齐)
def cformat(s): return s.upper().ljust(COLSIZ)

def setup():
    return RDBMSs[scanf('''
Choose a database system:

(M)ySQL
(G)adfly
(S)QLite

Enter choice: ''').strip().lower()[0]]

def connect(db, DBNAME='test'):
    global DB_EXC
    dbDir = '%s_%s' % (db, DBNAME)
    #sqlite是基于文件的（也可以使用：memory：作为文件名，从而在内存中创建数据库）
    if db == 'sqlite':
        try:
            import sqlite3
        except ImportError:
            try:
                from pysqlite2 import dbapi2 as sqlite3
            except ImportError:
                return None

        DB_EXC = sqlite3
        if not os.path.isdir(dbDir):
            os.mkdir(dbDir)
        cxn = sqlite3.connect(os.path.join(dbDir, DBNAME))

    elif db == 'mysql':
        try:
            # mysqlclient库也是mysqldb
            import MySQLdb
            import _mysql_exceptions as DB_EXC
            try:
                cxn = MySQLdb.connect(db=DBNAME)
            except DB_EXC.OperationalError:
                try:
                    cxn = MySQLdb.connect(user=DBUSER)
                    cxn.query('CREATE DATABASE %s' % DBNAME)
                    cxn.commit()
                    cxn.close()
                    cxn = MySQLdb.connect(db=DBNAME)
                except DB_EXC.OperationalError:
                    return None
        except ImportError:
            print('InterfaceError')
            try:
                import mysql.connector
                import mysql.connector.errors as DB_EXC
                try:
                    cxn = mysql.connector.Connect(
                        database=DBNAME,
                        user=DBUSER,
                        host='127.0.0.1',
                        password='',
                    )
                except DB_EXC.InterfaceError:
                    print('InterfaceError')
                    return None
            except ImportError:
                return None
    elif db == 'gadfly':
        try:
            from gadfly import gadfly
            DB_EXC = gadfly
        except ImportError:
            return None

        try:
            cxn = gadfly(DBNAME, dbDir)
        except IOError:
            cxn = gadfly()
            if not os.path.isdir(dbDir):
                os.mkdir(dbDir)
            cxn.startup(DBNAME, dbDir)
    else:
        return None
    return cxn

def create(cur):
    try:
        cur.execute(
            'CREATE TABLE users (login VARCHAR(%d),userid INTEGER,projid INTEGER)' % NAMELEN)
    except DB_EXC.ProgrammingError or DB_EXC.OperationalError:
        drop(cur)
        create(cur)

def drop(cur): return cur.execute('DROP TABLE users')

NAMES = (
    ('aaron', 8312), ('angela', 7603), ('dave', 7306),
    ('davina', 7902), ('elliot', 7911), ('ernie', 7410),
    ('jess', 7912), ('jim', 7512), ('larry', 7311),
    ('serena', 7003), ('stan', 7607), ('faye', 6812),
    ('amy', 7209), ('mona', 7404), ('jennifer', 7608),
)


def randName():
    pick = set(NAMES)
    while pick:
        yield pick.pop()

"""
SQLite和MySQL的适配器都是兼容DB-API的，所以他们的游标对象都存在executemany()函数，但是Gadfly就只能没吃插入一行
另一个差别：SQLite和Gadfly都使用的是qmark参数风格，而MySQL使用的是format参数风格。因此，格式化字符也存在一些差别
"""
def insert(cur, db):
    if db == 'sqlite':
        cur.executemany('INSERT INTO users VALUES(?,?,?)', [
                        (who, uid, rand(1.5)) for who, uid in randName()])
    elif db == 'gadfly':
        for who, uid in randName():
            cur.execute('INSERT INTO users VALUES(?,?,?)',
                        (who, uid, rand(1, 5)))
    elif db == 'mysql':
        cur.executemany('INSERT INTO users VALUES(%s,%s,%s)', [
                        (who, uid, rand(1, 5)) for who, uid in randName()])

#用于返回最后一次操作后影响的行数，不过如果游标对象不支持该属性（即不兼容DB-API）,则返回-1
def getRC(cur):return cur.rowcount if hasattr(cur, 'rowcount') else -1
#def getRC=lambda cur:cur.rowcount if hasattr(cur, 'rowcount') else -1
#py低于2.4
#getRC=lambda cur:(hasattr(cur,'rowcount') and [cur,rowcount] or [-1])[0]

def update(cur):
    fr = rand(1, 5)
    to = rand(1, 5)
    cur.execute('UPDATE users SET projid=%d WHERE projid=%d' % (to, fr))
    return fr, to, getRC(cur)


def delete(cur):
    rm = rand(1, 5)
    cur.execute('DELETE FROM users WHERE projid=%d' % rm)
    return rm, getRC(cur)

#从数据库拉取所有行，按照打印格式进行格式化
def dbDump(cur):
    cur.execute('SELECT * FROM users')
    getRC(cur)
    printf('\n%s' % ''.join(map(cformat, FIELDS)))
    """
    当我去除下面两行时，触发一个数据库错误：
        raise errors.InternalError("Unread result found")
    原因：
        原因是没有缓冲光标，结果被“延迟”加载，这意味着“fetchone”实际上只从查询的完整结果集中提取一行。
        当您再次使用相同的光标时，它会抱怨您仍然有等待获取的n-1个结果（其中n是结果集数量）。
        但是，当您使用缓冲游标时，连接器会在后台获取所有行，而您只需从连接器中获取一行，这样mysql数据库就不会抱怨。
    解决：
        cursor = cnx.cursor(buffered=True)
    """
    for data in cur.fetchall():
        printf(''.join(map(tformat, data)))


def main():
    db = setup()
    printf('*** Connect to %r database' % db)
    cxn = connect(db)
    if not cxn:
        printf('ERROR: %r not supported or unreachable,exit' % db)
        return
    cur = cxn.cursor()

    printf('\n*** Creating users table')
    create(cur)

    printf('\n*** Inserting names into table')
    insert(cur, db)
    dbDump(cur)

    printf('\n*** Randomly moving folks')
    fr, to, num = update(cur)
    printf('\r(%d users moved) from (%d) to (%d)' % (num, fr, to))
    dbDump(cur)

    printf('\n*** Randomly choosing group')
    rm, num = delete(cur)
    printf('\t(group #%d; %d users removed)' % (rm, num))
    dbDump(cur)

    printf('\n*** Dropping users table')
    drop(cur)
    printf('\n*** Close cxns')
    cur.close()
    cxn.commit()
    cxn.close()


if __name__ == '__main__':
    main()

ORM

对象关系映射（英语：(Object Relational Mapping，简称ORM，或O/RM，或O/R mapping），是一种程序技术，用于实现面向对象编程语言里不同类型系统的数据之间的转换 [1] 。从效果上说，它其实是创建了一个可在编程语言里使用的--“虚拟对象数据库”。

SQLAlchemy

安装

sudo pip3 install -U SQLAlchemy

声明层

#!/usr/bin/python3

from distutils.log import warn as printf
from os.path import dirname
from random import randrange as rand
from sqlalchemy import Column, Integer, String, create_engine, exc, orm
from sqlalchemy.ext.declarative import declarative_base
from ushuffle_db import DBNAME, NAMELEN, randName, FIELDS, tformat, cformat, setup

"""
兼容py2.x和py3.x版本的用户洗牌应用使用SQLAlchemy ORM 搭配后端数据库mysql或sqlite
"""
DSNs = {
    'mysql': 'mysql://root:passwd@localhost/%s' % DBNAME,
    'sqlite': 'sqlite:///:memory',
}

# 详请：http://docs.sqlalchemy.org/en/latest/orm/tutorial.html
Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'
    login = Column(String(NAMELEN))
    userid = Column(Integer, primary_key=True)
    projid = Column(Integer)

    def __str__(self):
        return ''.join(map(tformat, (self.login, self.userid, self.projid)))


class SQLALchemyTest(object):
    def __init__(self, dsn):
        try:
            eng = create_engine(dsn)
        except ImportError:
            raise RuntimeError()
        try:
            eng.connect()
        except exc.OperationalError:
            eng = create_engine(dirname(dsn))
            eng.execute('CREATE DATABASE %s' % DBNAME).close()
            eng = create_engine(dsn)

        Session = orm.sessionmaker(bind=eng)
        self.ses = Session()
        self.users = Users.__table__
        self.eng = self.users.metadata.bind = eng

    def insert(self):
        self.ses.add_all(
            Users(login=who, userid=userid, projid=rand(1, 5)) for who, userid in randName()
        )
        self.ses.commit()

    def update(self):
        fr = rand(1, 5)
        to = rand(1, 5)
        i = -1
        users = self.ses.query(Users).filter_by(projid=fr).all()
        for i, user in enumerate(users):
            user.projid = to
        self.ses.commit()
        return fr, to, i+1

    def delete(self):
        rm = rand(1, 5)
        i = -1
        users = self.ses.query(Users).filter_by(projid=rm).all()
        for i, user in enumerate(users):
            self.ses.delete(user)
        self.ses.commit()
        return rm, i+1

    def dbDump(self):
        printf('\n%s' % ''.join(map(cformat, FIELDS)))
        users = self.ses.query(Users).all()
        for user in users:
            printf(user)
        self.ses.commit()

    def __getattr__(self, attr):
        return getattr(self.users, attr)

    def finish(self):
        self.ses.connection().close()


def main():
    printf('*** Connect to %r database' % DBNAME)
    db = setup()
    if db not in DSNs:
        printf('\nError: %r not supported,exit' % db)
        return
    try:
        orm = SQLALchemyTest(DSNs[db])

    except RuntimeError:
        printf('\nERROR: %r not supported,exit' % db)
        return
    printf('\n*** Create users table (drop old one if appl.)')
    orm.drop(checkfirst=True)
    orm.create()

    printf('\n*** Insert names into table')
    orm.insert()
    orm.dbDump()

    printf('\n*** Move users to a random group')
    fr, to, num = orm.update()
    printf('\t(%d users moved) from (%d) to (%d)' % (num, fr, to))
    orm.dbDump()

    printf('\n*** Randomly delete group')
    rm, num = orm.delete()
    printf('\r(group #%d; %d users removed)' % (rm, num))
    orm.dbDump()

    printf('\n*** Drop users table')
    orm.drop()
    printf('\n*** Close cxns')
    orm.finish()
    
if __name__ == '__main__':
    main()

显式/“经典”的ORM访问(ushuffle_sae.py)

#!/usr/bin/python3

from distutils.log import warn as printf
from os.path import dirname
from random import randrange as rand
from sqlalchemy import Column, Integer, String, create_engine, exc, orm, MetaData, Table
from sqlalchemy.ext.declarative import declarative_base
from ushuffle_db import DBNAME, NAMELEN, randName, FIELDS, tformat, cformat, setup

DSNs = {
    'mysql': 'mysql://root@localhost/%s' % DBNAME,
    'sqlite': 'sqlite://:memory:',
}

class SQLAlchemyTest(object):
    def __init__(self, dsn):
        try:
            eng = create_engine(dsn)
        except ImportError:
            raise RuntimeError()

        try:
            cxn = eng.connect()
        except exc.OperationalError:
            try:
                eng = create_engine(dirname(dsn))   #mysql://root@localhost/
                eng.execute('CREATE DATABASE %s' % DBNAME).close()
                eng = create_engine(dsn)
                cxn = eng.connect()
            except exc.OperationalError:
                raise RuntimeError()

        metadata = MetaData()
        self.eng = metadata.bind = eng
        try:
            users = Table('users', metadata, autoload=True)
        except exc.NoSuchTableError:
            users = Table(
                'users', metadata,
                Column('login', String(NAMELEN)),
                Column('userid', Integer),
                Column('projid', Integer),
            )
        self.cxn=cxn
        self.users=users

    def insert(self):
        d=[dict(zip(FIELDS,[who,uid,rand(1,5)])) for who,uid in randName()]
        return self.users.insert().execute(*d).rowcount

    def update(self):
        users=self.users
        fr=rand(1,5)
        to=rand(1,5)
        return (fr,to,users.update(users.c.projid==fr).execute(projid=to).rowcount)

    def delete(self):
        users=self.users
        rm=rand(1,5)
        return (rm,users.delete(users.c.projid==rm).execute().rowcount)

    def dbDump(self):
        printf('\n%s'%''.join(map(cformat,FIELDS)))
        users=self.users.select().execute()
        for user in users.fetchall():
            printf(''.join(map(tformat,(user.login,user.userid,user.projid))))

        def __getattr__(self,attr):
            return getattr(self.users.attr)
        
        def finish(self):
            self.cxn.close()
    
def main():
    printf('\n*** Connect to %r database'%DBNAME)
    db=setup()
    if db not in DSNs:
        printf('\nERROR: %r not supported,exit'%db)
        return 
    
    try:
        orm=SQLAlchemyTest(DSNs(db))
    except RuntimeError:
        printf('\nERROR: %r not supported,exit'%db)
        return
    
    printf('\n*** Create users table (drop old one if appl.)')
    orm.drop(checkfirst=True)
    orm.create()

    printf('\n*** Insert name into table')
    orm.insert()
    orm.dbDump()

    printf('\n*** Move users to a random group')
    fr,to,num=orm.update()
    printf('\t(%d users moved) from (%d) to (%d)'%(num,fr,to))
    orm.dbDump()

    printf('\n*** Randomly delete group')
    rm,num=orm.delete()
    printf('\t(group #%d; %d users removed)'%(rm,num))

    printf('\n*** Drop users table')
    orm.drop()
    printf('\n*** Close cxns')
    orm.finish()

if __name__ == '__main__':
    main()

对比声明层代码与显示操作：

• 创建Table对象，而不是声明Base对象
• 没有使用session，而是执行单独的工作单元，进行自动提交，并没有事务性
• 使用Table对象进行所有数据库交互，而不是session Query

SQLObject

Python第一个主要的ORM

针对ushuffle_db.py与ushuffle_sae.py移植版

ushuffle_so.py

#!/usr/bin/python3

from distutils.log import warn as printf
from os.path import dirname
from sqlobject import *
from random import randrange as rand
from ushuffle_db import DBNAME,NAMELEN,randName,FIELDS,tformat,cformat,setup
"""
兼容python2/3的用户洗牌应用使用的SQLObject ORM搭配后端数据库Mysql/sqlite
"""

DSNs={
    'mysql':'mysql://root@localhost/%s'%DBNAME,
    'sqlite':'sqlite:///:memory:',
}

class Users(SQLObject):
    login=StringCol(length=NAMELEN)
    userid=IntCol()
    projid=IntCol()
    def __str__(self):
        return ''.join(map(tformat,(self.login,self.userid,self.projid)))
    
class SQLObjectTest(object):
    def __init__(self,dsn):
        try:
            cxn=connectionForURI(dsn)
        except ImportError:
            raise RuntimeError()
        try:
            cxn.releaseConnection(cxn.getConnection())
        except dberrors.OperationalError:
            cxn=connectionForURI(dirname(dsn))
            cxn.query('CREATE DATABASE %s'%DBNAME)
            cxn=connectionForURI(dsn)
        self.cxn=sqlhub.processConnection=cxn

    def insert(self):
        for who,userid in randName():
            Users(login=who,userid=userid,projid=rand(1,5))
    
    def update(self):
        fr=rand(1,5)
        to=rand(1,5)
        users=Users.selectBy(projid=fr)
        for i,user in enumerate(users):
            users.projid=to
        return fr,to,i+1
    
    def delete(self):
        rm=rand(1,5)
        users=Users.selectBy(projid=rm)
        for i,user in enumerate(users):
            user.destroySelf()
        return rm,i+1
    
    def dbDump(self):
        printf('\n%s'%''.join(map(cformat,FIELDS)))
        for user in Users.select():
            printf(user)

    def finish(self):
        self.cxn.close()

def main():
    printf('*** Connect to %r database'%DBNAME)
    db=setup()
    if db not in DSNs:
        printf('\nERROR: %r not support,exit'%db)
        return 
    try:
        orm=SQLObjectTest(DSNs[db])
    except RuntimeError:
        printf('\nERROR: %r not support,exit'%db)
        return
    
    printf('\n*** Create users table (drop old one if appl.)')
    Users.dropTable(True)
    Users.createTable()

    printf('\n*** Insert names into table')
    orm.insert()
    orm.dbDump()

    printf('\n*** Move users to a random group')
    fr,to,num=orm.update()
    printf('\t(%d users moved) from (%d) to (%d)'%(num,fr,to))
    orm.dbDump()

    printf('\n*** Randomly delete group')
    rm,num=orm.delete()
    printf('\t(group #%d; %d users removed)'%(rm,num))
    orm.dbDump()

    printf('\n *** Drop users table')
    Users.dropTable()
    printf('\n*** Close cxns')
    orm.finish()

if __name__=='__main__':
    main()

非关系数据库

NoSQL

MongoDB(ushuffle_mongo.py)

#!/usr/bin/python3

from distutils.log import warn as printf
from random import randrange as rand
from pymongo import MongoClient, errors
from ushuffle_db import DBNAME, randName, FIELDS, tformat, cformat
"""
启动mongo：
    - service mongod start
    - mongod (查看mongo服务情况)
    - mongo
"""

COLLECTION = 'users'


class MongoTest(object):
    def __init__(self):
        try:
            self.cxn = MongoClient() #连接mongo服务
        except errors.AutoReconnect:
            raise RuntimeError()
        self.db = self.cxn[DBNAME] #连接指定数据库
        self.users = self.db[COLLECTION]

    def insert(self):
        self.users.insert(
            dict(login=who, userid=uid, projid=rand(1, 5)) for who, uid in randName()
        )

    def update(self):
        fr=rand(1,5)
        to=rand(1,5)
        i=-1
        for i,user in enumerate(self.users.find({'projid':fr})):
            self.users.update(user,{'$set':{'projid':to}})
        return fr,to,i+1

    def delete(self):
        rm=rand(1,5)
        i=-1
        for i,user in enumerate(self.users.find({'projid':rm})):
            self.users.remove(user)
        return rm,i+1
    
    def dbDump(self):
        printf('\n%s'%''.join(map(cformat,FIELDS)))
        for user in self.users.find():
            printf(''.join(map(tformat,(user[k.lower()] for k in FIELDS))))
            #printf(''.join(map(tformat,user)))

    """
    The “disconnect” method is removed
    - client.disconnect()
    can be replaced by this with PyMongo 2.9 or later:
    - client.close()
    """
    def finish(self):
        self.cxn.close()
    
def main():
    printf('*** Connect to %r database'%DBNAME)
    try:
        mongo=MongoTest()
    except RuntimeError:
        printf('\nERROR: MongoDB server unreachable,exit')
        return
    printf('\n*** Insert names into table')
    mongo.insert()
    mongo.dbDump()

    printf('\n*** Move users to a random group')
    fr,to,num=mongo.update()
    printf('\t(%d users moved) from (%d) to (%d)'%(num,fr,to))
    mongo.dbDump()

    printf('\n*** Randomly delete group')
    rm,num=mongo.delete()
    printf('\t(group #%d;%d users removed)'%(rm,num))
    mongo.dbDump()

    printf('\n*** Drop users table')
    mongo.db.drop_collection(COLLECTION)
    printf('\n*** Close cxns')
    mongo.finish()

if __name__=='__main__':
    main()

本文由 weti 创作，采用 知识共享署名4.0 国际许可协议进行许可
本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名
最后编辑时间为: Aug 21, 2018 at 12:21 am