python类Blueprint()的实例源码

def add_swagger_api_route(app, target_route, swagger_json_route):
    """
    mount a swagger statics page.
    app: the sanic app object
    target_route: the path to mount the statics page.
    swagger_json_route: the path where the swagger json definitions is
                        expected to be.
    """
    static_root = get_swagger_static_root()
    swagger_body = generate_swagger_html(
        STATIC_ROOT, swagger_json_route
    ).encode("utf-8")

    async def swagger_ui(request):
        return HTTPResponse(body_bytes=swagger_body, content_type="text/html")

    bp = Blueprint('swagger')
    bp.static(STATIC_ROOT, static_root)

    app.add_route(swagger_ui, target_route, methods=["GET"])
    app.blueprint(bp)

def init_databases(self):
        """Initialize database connection pools from settings.py, 
        setting up Sanic blueprints for server start and stop."""
        for database in settings.DATABASES:
            db_blueprint = Blueprint(f'{self.name}_db_blueprint_{database}')
            connection_settings = settings.DATABASES[database].copy()
            if not connection_settings['user']:
                connection_settings.pop('user')
                connection_settings.pop('password')
            @db_blueprint.listener('before_server_start')
            async def setup_connection_pool(app, loop):
                self._db_pools[database] = await create_pool(**connection_settings)
            @db_blueprint.listener('after_server_stop')
            async def close_connection_pool(app, loop):
                if database in self._db_pools and self._db_pools[database]:
                    await self._db_pools[database].close()
            self.server.blueprint(db_blueprint)

def test_bp_combined_limit(self):
        app, limiter = self.build_app(config={}, global_limits=['1/day'])
        bp = Blueprint('/bp')
        limiter.limit('1 per hour')(bp)

        @bp.route("/bp1")
        @limiter.limit('2 per hour')
        async def bp_t1(request):
            return text("bp_t1")

        app.blueprint(bp)

        cli = app.test_client
        self.assertEqual(200, cli.get("/bp1")[1].status)
        self.assertEqual(200, cli.get("/bp1")[1].status)
        self.assertEqual(429, cli.get("/bp1")[1].status)

def init_smtp(self):
        """Initialize smtp connection"""
        if not 'SMTP' in settings:
            return
        smtp_blueprint = Blueprint(f'{self.name}_smtp_blueprint')
        @smtp_blueprint.listener('before_server_start')
        async def connect_smtp(app, loop):
            if settings.SMTP.get('ssl', False):
                self._smtp = SMTP_SSL(hostname=settings.SMTP['host'], port=settings.SMTP['port'])
            else:
                self._smtp = SMTP(hostname=settings.SMTP['host'], port=settings.SMTP['port'])
            await self._smtp.connect()
            if 'username' in settings.SMTP and 'password' in settings.SMTP:
                await self._smtp.auth.auth(settings.SMTP['username'], settings.SMTP['password'])
        self.server.blueprint(smtp_blueprint)

def test_bp_limit(self):
        app, limiter = self.build_app(config={}, global_limits=['1/day'])
        bp = Blueprint('/bp')
        limiter.limit('2 per hour')(bp)

        @bp.route("/bp1")
        async def bp_t1(request):
            return text("bp_t1")

        app.blueprint(bp)

        cli = app.test_client
        self.assertEqual(200, cli.get("/bp1")[1].status)
        self.assertEqual(200, cli.get("/bp1")[1].status)
        self.assertEqual(429, cli.get("/bp1")[1].status)