python中bottle使用实例代码

作者:qq_26372385

这篇文章主要介绍了python bottle使用实例,主要是搞懂python中如何来组织包,如何调用包,如何读取配置文件,连接数据库,设置路由,路由分组,具体示例代码跟随小编一起看看吧

模仿学习同事的代码来写的,主要是搞懂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 = 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/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使用内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
  • Python写代码的七条重要技巧介绍
    Python写代码的七条重要技巧介绍
    2021-12-12
  • Python实现如何根据文件后缀进行分类
    Python实现如何根据文件后缀进行分类
    2021-12-12
  • Python matplotlib 绘制散点图详解建议收藏
    Python matplotlib 绘制散点图详解建议收藏
    2021-12-12
  • Python传统图像处理之皮肤区域检测详解
    Python传统图像处理之皮肤区域检测详解
    2021-12-12
  • Python Pyqt5多线程更新UI代码实例(防止界面卡死)
    Python Pyqt5多线程更新UI代码实例(防止界面卡死)
    2021-12-12
  • 详解Python函数中的几种参数
    详解Python函数中的几种参数
    2021-12-12
  • Python解压可迭代对象赋值给多个变量详解
    Python解压可迭代对象赋值给多个变量详解
    2021-12-12
  • Python爬虫采集微博视频数据
    Python爬虫采集微博视频数据
    2021-12-12
  • 美国设下计谋,用娘炮文化重塑日本,已影响至中国
    美国设下计谋,用娘炮文化重塑日本,已影响至中国
    2021-11-19
  • 时空伴随者是什么意思?时空伴随者介绍
    时空伴随者是什么意思?时空伴随者介绍
    2021-11-09
  • 工信部称网盘企业免费用户最低速率应满足基本下载需求,天翼云盘回应:坚决支持,始终
    工信部称网盘企业免费用户最低速率应满足基本下载需求,天翼云盘回应:坚决支持,始终
    2021-11-05
  • 2022年放假安排出炉:五一连休5天 2022年所有节日一览表
    2022年放假安排出炉:五一连休5天 2022年所有节日一览表
    2021-10-26
  • 电脑版 - 返回首页

    2006-2023 脚本之家 JB51.Net , All Rights Reserved.
    苏ICP备14036222号