python中bottle使用实例代码
模仿学习同事的代码来写的,主要是搞懂python中如何来组织包,如何调用包,如何读取配置文件,连接数据库,设置路由,路由分组。(注:使用的是python3.6)
整体目录设计如下:
根据调用层级从上往下来说:
首先项目根目录下的main.py是整个程序的入口,主要作用启动http服务器,调用分组路由。
main.py
import bottle from confg.conf import conf from api.user import user db_url = conf.db.url default_app = bottle.default_app() #相当于分组路由 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下的文件。(注:一般来说controller层,service层是同级的,本项目其实api下的非service文件都是属于controller层,所以还是同一层的,因为要遵守调用顺序,不然可能会发生循环调用)。
/api/user.py文件
import logging from bottle import request #db数据库引擎 from common.base import db #调用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 #相当于分组路由 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层也就是数据库操作层(但是明显虽然有dao层但是数据库操作的逻辑已经在service层里了)
最后是读取配置文件和创建数据库引擎。
读取配置文件使用的包是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) # 声明参数选项 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='数据库连接池是否记录 checkouts/checkins操作'), cfg.intopt('pool_size', default=1000, help='数据库连接池中保持打开的连接数量'), cfg.intopt('pool_recycle', default=600, help='数据库连接池在连接被创建多久(单位秒)以后回收连接') ] token_opts = [ cfg.stropt('project_domain_name'), cfg.stropt('project_name'), ] cinder_opts = (auth_opts + token_opts) mysqlcinder_opts = (mysql_opts) # 注册参数选项 conf = cfg.conf # 注册组 conf.register_group(opt_group) conf.register_group(mysql_opt_group) # 将各个选项注册进组里 conf.register_opts(cinder_opts, group=opt_group) conf.register_opts(mysqlcinder_opts, group=mysql_opt_group) if __name__ == "__main__": # 要读取哪个配置文件 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
它的使用方法是,先声明参数选项就是(相当于声明组)
mysql_opt_group = cfg.optgroup('db'),
然后声明组内的选项,
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='数据库连接池是否记录 checkouts/checkins操作'), cfg.intopt('pool_size', default=1000, help='数据库连接池中保持打开的连接数量'), cfg.intopt('pool_recycle', default=600, help='数据库连接池在连接被创建多久(单位秒)以后回收连接') ]
拼接选项
mysqlcinder_opts = (mysql_opts)
接着注册组,
conf.register_group(mysql_opt_group)
最后将选项注册进组。
conf.register_opts(mysqlcinder_opts, group=mysql_opt_group)
当然最重要的注册参数选项,我的理解就是暴露句柄。
# 注册参数选项
conf = cfg.conf
然后创建数据库引擎
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,那么每次从连接池中拿连接的时候,都会向数据库发送一个类似 # select 1的测试查询语句来判断服务器是否正常运行。当该连接出现disconnect的情况时, # 该连接连同pool中的其它连接都会被回收 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]
这里引用配置文件的数据,直接引入conf
from confg import conf
然后使用
conf.db.echo_pool
创建句柄,
与我之前使用的方法不同的是,这里的数据库引擎不需要在使用的地方引入了,会在main里注册路由分组时,通过plugin插件自动将数据库引擎导入。这也是我有点搞不懂的地方,虽然更方便,但是不知道就很难知道了,问了同事才知道是怎么回事。
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就是相当与golang的中间件,不过作用范围是全部路由。
这里创建数据库句柄并使用是一个比较绕的过程。总体思路:
1.写一个bottle plugin,创建数据库句柄,然后install安装这个plugin。就可以在所有的路由中自动引入这个插件(就是不用在包里在导入db句柄了,bottle会自动导入)。
/common/base.py 创建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): """ 创建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就不需要导入了,可以直接使用。
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="创建时间") updated_at = column("updated_at", datetime, comment="更新时间") deleted_at = column("deleted_at", datetime, comment="删除时间") 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提供时间戳基类""" created_at = column('created_at', timestamp(true), default=datetime.now, comment=u"创建时间") updated_at = column('updated_at', timestamp(true), default=datetime.now, onupdate=datetime.now, comment=u"更新时间")
到此这篇关于python bottle使用实例的文章就介绍到这了,更多相关python bottle使用内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!