模仿學(xué)習(xí)同事的代碼來(lái)寫的，主要是搞懂python中如何來(lái)組織包，如何調(diào)用包，如何讀取配置文件，連接數(shù)據(jù)庫(kù)，設(shè)置路由，路由分組。（注：使用的是python3.6）
整體目錄設(shè)計(jì)如下：
根據(jù)調(diào)用層級(jí)從上往下來(lái)說(shuō)：
首先項(xiàng)目根目錄下的main.py是整個(gè)程序的入口，主要作用啟動(dòng)http服務(wù)器，調(diào)用分組路由。
main.py
import bottle
from confg.conf import CONF
from api.user import User
db_url = CONF.db.url
default_app = bottle.default_app()
#相當(dāng)于分組路由
default_app.mount("/user", User(db_url, "").app)
app = default_app
if __name__ == '__main__':
bottle.run(app=app,
host="localhost",
port="8000")
接著是controller層，就是api目錄。api目錄包括service文件夾和api下的文件。（注：一般來(lái)說(shuō)controller層，service層是同級(jí)的，本項(xiàng)目其實(shí)api下的非service文件都是屬于controller層，所以還是同一層的，因?yàn)橐袷卣{(diào)用順序，不然可能會(huì)發(fā)生循環(huán)調(diào)用）。
/api/user.py文件
import logging
from bottle import request
#db數(shù)據(jù)庫(kù)引擎
from common.base import DB
#調(diào)用service層
from api.service.user import UserService
logger = logging.getLogger("arview")
class User(DB, UserService):
def __init__(self, *args, **kwargs):
print(">>> User init begin")
logging.debug('>>> User init begin')
super(User, self).__init__(*args, **kwargs)
self.dispatch()
logger.debug('>>> User init end')
def create(self, db=None):
create_body = request.json
create_data = self.create_user(create_body, db)
return create_data
def delete(self, db=None):
delete_body = request.json
delete_data = self.delete_user(delete_body, db)
return delete_data
def list(self, db=None):
list_data = self.list_user(db)
return list_data
#相當(dāng)于分組路由
def dispatch(self):
self.app.route('/listUser', method='post')(self.list)
self.app.route('/createUser', method='post')(self.create)
self.app.route('/deleteUser', method='post')(self.delete)
/service/user.py
import time
#model層
from db.models.user import UserModel
class UserService(object):
def list_user(self, db):
user_info_list = db.query(UserModel).all()
for item in user_info_list:
print(item.username)
return user_info_list
def create_user(self, create_body, db):
user_model = UserModel(
username=create_body.get("username"),
password=create_body.get("password"),
role=create_body.get("role"),
create_time=time.time()
)
db.add(user_model)
db.commit()
return "success"
def delete_user(self, delete_body, db):
db.query(UserModel).filter(UserModel.id == (delete_body["id"])).delete()
db.commit()
return delete_body
然后是dao層也就是數(shù)據(jù)庫(kù)操作層（但是明顯雖然有dao層但是數(shù)據(jù)庫(kù)操作的邏輯已經(jīng)在service層里了）
最后是讀取配置文件和創(chuàng)建數(shù)據(jù)庫(kù)引擎。
讀取配置文件使用的包是oslo_config。
conf.py
# coding:utf8
# from __future__ import print_function
from oslo_config import cfg
DEFAULT_ARVIEW_DB_NAME = 'ginblog'
DEFAULT_ARVIEW_DB_USER = 'root'
DEFAULT_ARVIEW_DB_USER_PASSWORD = '33demysql'
DEFAULT_ARVIEW_DB_HOST = '81.68.179.136'
DEFAULT_ARVIEW_DB_PORT = 3306
DEFAULT_ARVIEW_DB_URL_TEMPLATE = 'mysql+mysqlconnector://{}:{}@' \
'{}:{}/{}?charset=utf8'
DEFAULT_ARVIEW_DB_URL = DEFAULT_ARVIEW_DB_URL_TEMPLATE.format(
DEFAULT_ARVIEW_DB_USER,
DEFAULT_ARVIEW_DB_USER_PASSWORD,
DEFAULT_ARVIEW_DB_HOST,
DEFAULT_ARVIEW_DB_PORT,
DEFAULT_ARVIEW_DB_NAME)
# 聲明參數(shù)選項(xiàng)
opt_group = cfg.OptGroup('keystone_authtoken')
mysql_opt_group = cfg.OptGroup('db')
auth_opts = [
cfg.StrOpt('memcached_servers',
default='localhost:11211',
choices=("localhost:11211", "0.0.0.0:11211"),
help=('localhost local', '0.0.0.0 So listen')
),
cfg.StrOpt('signing_dir',
default='/var/cache/cinder',
choices=("/var/cache/cinder", "/var/cache/cinder"),
),
]
# mysql
mysql_opts = [
cfg.StrOpt('url', default=DEFAULT_ARVIEW_DB_URL),
cfg.StrOpt('Db', default='3mysql'),
cfg.StrOpt('DbHost', default='381.68.179.136'),
cfg.StrOpt('DbPort', default='33306'),
cfg.StrOpt('DbUser', default='3DbUser'),
cfg.StrOpt('DbPassWord', default='3DbPassWord'),
cfg.StrOpt('DbName', default='3DbName'),
cfg.BoolOpt('create', default=False),
cfg.BoolOpt('commit', default=True),
cfg.BoolOpt('echo', default=True, help='是否顯示回顯'),
cfg.BoolOpt('echo_pool', default=False, help='數(shù)據(jù)庫(kù)連接池是否記錄 checkouts/checkins操作'),
cfg.IntOpt('pool_size', default=1000, help='數(shù)據(jù)庫(kù)連接池中保持打開的連接數(shù)量'),
cfg.IntOpt('pool_recycle', default=600, help='數(shù)據(jù)庫(kù)連接池在連接被創(chuàng)建多久（單位秒）以后回收連接')
]
token_opts = [
cfg.StrOpt('project_domain_name'),
cfg.StrOpt('project_name'),
]
CINDER_OPTS = (auth_opts +
token_opts)
MYSQLCINDER_OPTS = (mysql_opts)
# 注冊(cè)參數(shù)選項(xiàng)
CONF = cfg.CONF
# 注冊(cè)組
CONF.register_group(opt_group)
CONF.register_group(mysql_opt_group)
# 將各個(gè)選項(xiàng)注冊(cè)進(jìn)組里
CONF.register_opts(CINDER_OPTS, group=opt_group)
CONF.register_opts(MYSQLCINDER_OPTS, group=mysql_opt_group)
if __name__ == "__main__":
# 要讀取哪個(gè)配置文件
CONF(default_config_files=['cinder.conf'])
print('mysql Db配置組為%s' % (CONF.db.Db))
print('mysql DbHost%s' % (CONF.db.DbHost))
print('mysql DbPort配置組為%s' % (CONF.db.DbPort))
print('mysql DbUser%s' % (CONF.db.DbUser))
配置文件cinder.conf
[db]
Db = mysql
DbHost = 81.68.179.136
DbPort = 3306
DbUser = root
DbPassWord = 33demysql
DbName = ginblog
create = false
commit = true
echo = false
echo_pool = false
pool_size = 1000
pool_recycle =600
它的使用方法是，先聲明參數(shù)選項(xiàng)就是(相當(dāng)于聲明組)
mysql_opt_group = cfg.OptGroup('db')，
然后聲明組內(nèi)的選項(xiàng)，
mysql_opts = [
cfg.StrOpt('url', default=DEFAULT_ARVIEW_DB_URL),
cfg.StrOpt('Db', default='3mysql'),
cfg.StrOpt('DbHost', default='381.68.179.136'),
cfg.StrOpt('DbPort', default='33306'),
cfg.StrOpt('DbUser', default='3DbUser'),
cfg.StrOpt('DbPassWord', default='3DbPassWord'),
cfg.StrOpt('DbName', default='3DbName'),
cfg.BoolOpt('create', default=False),
cfg.BoolOpt('commit', default=True),
cfg.BoolOpt('echo', default=True, help='是否顯示回顯'),
cfg.BoolOpt('echo_pool', default=False, help='數(shù)據(jù)庫(kù)連接池是否記錄 checkouts/checkins操作'),
cfg.IntOpt('pool_size', default=1000, help='數(shù)據(jù)庫(kù)連接池中保持打開的連接數(shù)量'),
cfg.IntOpt('pool_recycle', default=600, help='數(shù)據(jù)庫(kù)連接池在連接被創(chuàng)建多久（單位秒）以后回收連接')
]
拼接選項(xiàng)
MYSQLCINDER_OPTS = (mysql_opts)
接著注冊(cè)組，
CONF.register_group(mysql_opt_group)
最后將選項(xiàng)注冊(cè)進(jìn)組。
CONF.register_opts(MYSQLCINDER_OPTS, group=mysql_opt_group)
當(dāng)然最重要的注冊(cè)參數(shù)選項(xiàng)，我的理解就是暴露句柄。
# 注冊(cè)參數(shù)選項(xiàng)
CONF = cfg.CONF
然后創(chuàng)建數(shù)據(jù)庫(kù)引擎
common/utils/sqlalchemy_util.py
import logging
from json import loads as json_loads
from sqlalchemy.engine import create_engine
from sqlalchemy.pool import QueuePool
from confg import CONF
SQLALCHEMY_ENGINE_CONTAINER = {}
logger = logging.getLogger("arview")
def json_deserializer(s, **kw):
if isinstance(s, bytes):
return json_loads(s.decode('utf-8'), **kw)
else:
return json_loads(s, **kw)
def get_sqlalchemy_engine(db_url):
if db_url not in SQLALCHEMY_ENGINE_CONTAINER:
engine = create_engine(db_url, echo=CONF.db.echo,
# pool_pre_ping如果值為True，那么每次從連接池中拿連接的時(shí)候，都會(huì)向數(shù)據(jù)庫(kù)發(fā)送一個(gè)類似
# select 1的測(cè)試查詢語(yǔ)句來(lái)判斷服務(wù)器是否正常運(yùn)行。當(dāng)該連接出現(xiàn)disconnect的情況時(shí)，
# 該連接連同pool中的其它連接都會(huì)被回收
pool_pre_ping=True,
echo_pool=CONF.db.echo_pool,
pool_size=CONF.db.pool_size,
pool_recycle=CONF.db.pool_recycle,
json_deserializer=json_deserializer,
poolclass=QueuePool)
logger.info('Create sqlalchemy engine %s', engine)
SQLALCHEMY_ENGINE_CONTAINER[db_url] = engine
return SQLALCHEMY_ENGINE_CONTAINER[db_url]
這里引用配置文件的數(shù)據(jù)，直接引入CONF
from confg import CONF
然后使用
CONF.db.echo_pool
創(chuàng)建句柄，
與我之前使用的方法不同的是，這里的數(shù)據(jù)庫(kù)引擎不需要在使用的地方引入了，會(huì)在main里注冊(cè)路由分組時(shí)，通過(guò)plugin插件自動(dòng)將數(shù)據(jù)庫(kù)引擎導(dǎo)入。這也是我有點(diǎn)搞不懂的地方，雖然更方便，但是不知道就很難知道了，問(wèn)了同事才知道是怎么回事。
bottle源碼
def install(self, plugin):
''' Add a plugin to the list of plugins and prepare it for being
applied to all routes of this application. A plugin may be a simple
decorator or an object that implements the :class:`Plugin` API.
'''
plugin就是相當(dāng)與golang的中間件，不過(guò)作用范圍是全部路由。
這里創(chuàng)建數(shù)據(jù)庫(kù)句柄并使用是一個(gè)比較繞的過(guò)程?？傮w思路：
1.寫一個(gè)bottle plugin，創(chuàng)建數(shù)據(jù)庫(kù)句柄，然后install安裝這個(gè)plugin。就可以在所有的路由中自動(dòng)引入這個(gè)插件（就是不用在包里在導(dǎo)入db句柄了，bottle會(huì)自動(dòng)導(dǎo)入）。
/common/base.py 創(chuàng)建plugin并安裝
import logging
from bottle import Bottle
from confg.conf import CONF
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from db.models.base import Base as ApiModelBase
from common.utils.sqlalchemy_util import get_sqlalchemy_engine
from bottle_sqlalchemy import SQLAlchemyPlugin
logger = logging.getLogger("arview")
base = ApiModelBase # sqlalchemy orm base class
class Plugins:
SQLALCHEMY_PLUGIN = None # sqlalchemy plugin, global only one instance
APSCHEDULER_PLUGIN = None # apsechduler plugin. global only one instance
class Base(object):
def __init__(self, *args, **kwargs):
logger.debug('>>>> Base init begin')
self.app = Bottle()
# self.app.install(SwaggerPlugin(self._type))
logger.debug('>>>> Base init end')
class DB(Base):
def __init__(self, db_url, create=None, commit=None, *args, **kwargs):
print('db_url:', db_url)
super(DB, self).__init__(*args, **kwargs)
if create is None:
create = CONF.db.create
if commit is None:
commit = CONF.db.commit
if Plugins.SQLALCHEMY_PLUGIN is None:
Plugins.SQLALCHEMY_PLUGIN = _create_sqlalchemy_plugin(db_url, create=create, commit=commit)
self.app.install(Plugins.SQLALCHEMY_PLUGIN)
logger.debug("Install plugin: sqlalchemy.")
# if CONF.api.enable_request_interval_plugin:
# self.app.install(RequestTimeIntervalPlugin())
logger.debug('>>>> DB init end')
class CommonBase(object):
def __init__(self):
self._db = None
@property
def db(self):
if not self._db:
DBURL = "mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8".format(CONF.mysql.DbUser,
CONF.mysql.DbPassWord,
CONF.mysql.DbHost,
CONF.mysql.DbPort,
CONF.mysql.DbName)
engine = create_engine(DBURL, echo=False)
self._db = sessionmaker()(bind=engine)
return self._db
@db.deleter
def db(self):
if self._db:
self._db.commit()
self._db.close()
self._db = None
def _create_sqlalchemy_plugin(db_url, create, commit):
"""
創(chuàng)建sqlalchemy插件
:param db_url:
:param echo:
:param create:
:param commit:
:return:
"""
logger.debug('>>>> create sqlalchemy plugin begin')
engine = get_sqlalchemy_engine(db_url)
plugin = SQLAlchemyPlugin(engine, metadata=ApiModelBase.metadata, create=create, commit=commit, use_kwargs=True)
logger.debug('>>>> create sqlalchemy plugin %s' % plugin)
return plugin
最后使用
/api/user.py
import logging
from bottle import request
from common.base import DB
from api.service.user import UserService
logger = logging.getLogger("arview")
class User(DB, UserService):
def __init__(self, *args, **kwargs):
print(">>> User init begin")
logging.debug('>>> User init begin')
super(User, self).__init__(*args, **kwargs)
self.dispatch()
logger.debug('>>> User init end')
def create(self, db=None):
create_body = request.json
create_data = self.create_user(create_body, db)
return create_data
def delete(self, db=None):
delete_body = request.json
delete_data = self.delete_user(delete_body, db)
return delete_data
def list(self, db=None):
list_data = self.list_user(db)
return list_data
def dispatch(self):
self.app.route('/listUser', method='post')(self.list)
self.app.route('/createUser', method='post')(self.create)
self.app.route('/deleteUser', method='post')(self.delete)
這里的db就不需要導(dǎo)入了，可以直接使用。
db層
主要是模型層 /db/model/user.py
from sqlalchemy import Column, String, Enum, TIMESTAMP, Boolean, Integer, BIGINT, DATETIME
from db.models.base import Base
class UserModel(Base):
__tablename__ = "user"
id = Column("id", BIGINT, primary_key=True, comment="用戶id")
created_at = Column("created_at", DATETIME, comment="創(chuàng)建時(shí)間")
updated_at = Column("updated_at", DATETIME, comment="更新時(shí)間")
deleted_at = Column("deleted_at", DATETIME, comment="刪除時(shí)間")
username = Column("username", String(20), comment="用戶名")
password = Column("password", String(500), comment="密碼")
role = Column("role", BIGINT, comment="角色")
def __init__(self, id, created_at, updated_at, deleted_at, username, password, role):
self.id = id
self.created_at = created_at
self.updated_at = updated_at
self.deleted_at = deleted_at
self.username = username
self.password = password
self.role = role
/db/model/base.py
from datetime import datetime
from sqlalchemy import Column, TIMESTAMP
from sqlalchemy.ext.declarative import declarative_base
# sqlalchemy orm base class
Base = declarative_base()
class TimestampMixin(object):
"""為ORM提供時(shí)間戳基類"""
created_at = Column('created_at', TIMESTAMP(True), default=datetime.now,
comment=u"創(chuàng)建時(shí)間")
updated_at = Column('updated_at', TIMESTAMP(True), default=datetime.now,
onupdate=datetime.now, comment=u"更新時(shí)間")
到此這篇關(guān)于python bottle使用實(shí)例的文章就介紹到這了,更多相關(guān)python bottle使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:- Python使用Py2neo創(chuàng)建Neo4j的節(jié)點(diǎn)、關(guān)系及路徑
- Python使用py2neo操作圖數(shù)據(jù)庫(kù)neo4j的方法詳解
- python利用文件讀寫編寫一個(gè)博客
- 手把手帶你用python爬取小姐姐私房照
- Python time.time()方法
- Python接口自動(dòng)化之接口依賴
- python使用py2neo查詢Neo4j的節(jié)點(diǎn)、關(guān)系及路徑
