Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from flask import Flask, request
from flasgger import Swagger
from nameko.standalone.rpc import ClusterRpcProxy
from flasgger.utils import swag_from
import logging
import os
app = Flask(__name__)
Swagger(app)
CONFIG = {'AMQP_URI': "pyamqp://guest:guest@localhost"}
@app.route('/send', methods=['POST'])
@swag_from('docs/send.yml')
def send():
logger = app.logger
type = request.json.get('type')
body = request.json.get('body')
address = request.json.get('address')
logger.info('Get message: %s,%s,%s' % (type,body,address))
with ClusterRpcProxy(CONFIG) as rpc:
# asynchronously spawning and email notification
rpc.yowsup.send(type,body,address)
#
backend.autocommit = True
# --------------------------------------------------------------------------
# Routes
# --------------------------------------------------------------------------
# Catalog
routes_catalog(app)
app.config['APP_DB'] = backend
# --------------------------------------------------------------------------
# Enable doc?
# --------------------------------------------------------------------------
if args.NOD_DOC is False:
Swagger(app)
app.run(host=args.IP,
port=args.PORT)
def create_app(dev=True):
"""
Creates Flask instance & initialize
:rtype: Flask
"""
print('[INFO] Flask application initialized with {} mode.'.format('DEV' if dev else 'PRODUCTION'))
app_ = Flask(__name__)
app_.config.from_object(DevConfig if dev else ProductionConfig)
JWTManager().init_app(app_)
CORS().init_app(app_)
Swagger(app_, template=TEMPLATE)
Mongo().init_app(app_)
Router().init_app(app_)
return app_
# -*- coding: utf-8 -*-
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
from flask_ldapconn import LDAPConn
from flasgger import Swagger
from flask_migrate import Migrate
cors = CORS()
ldap = LDAPConn()
swagger = Swagger()
db = SQLAlchemy()
migrate = Migrate()
import calendar
from flasgger import Swagger
websocket_url = "ws://localhost:8090/ws"
ws = create_connection(websocket_url)
app = Flask(__name__)
from flask_cors import CORS, cross_origin
CORS(app)
app.config['SWAGGER'] = {
'title': 'BTS UDF Wrapper',
'uiversion': 2
}
Swagger(app, template_file='wrapper.yaml')
# postgres
import psycopg2
postgres_host = 'localhost'
postgres_database = 'explorer'
postgres_username = 'postgres'
postgres_password = 'posta'
# end postgres
if __name__ == "__main__":
app.run(host='0.0.0.0', port=5001)
@app.route('/config')
def config():
results = {}
# coding: utf-8
"""
This tests the use of a view coming from installed
package.
"""
from flask import Flask, jsonify
from flasgger import Swagger
from flasgger_package import package_view
app = Flask(__name__)
swag = Swagger(app)
app.add_url_rule(
'/v1/decorated/',
view_func=package_view
)
@app.route('/v2/decorated/')
def package_view_2(username):
"""
This is the summary defined in yaml file
First line is the summary
All following lines until the hyphens is added to description
the format of the first lines until 3 hyphens will be not yaml compliant
but everything below the 3 hyphens should be.
---
"""
Another example using Marshmallow schemas in SwaggerView (MethodView)
"""
# coding: utf-8
from flask import Flask, jsonify, request
from flasgger import Schema, Swagger, SwaggerView, fields
app = Flask(__name__)
app.config['SWAGGER'] = {
"title": "API using Marshmallow",
"uiversion": 2
}
swag = Swagger(app)
class User(Schema):
username = fields.Str(required=True)
age = fields.Int(required=True, min=18)
tags = fields.List(fields.Str())
class UserPostView(SwaggerView):
parameters = User
# parameters = [
# {
# 'name': 'body',
# 'in': 'body',
if __name__ != "__main__":
return f(*args, **kwargs)
auth = request.authorization
if not auth or not check_auth(auth.username, auth.password):
return authenticate()
return f(*args, **kwargs)
return decorated
app = Flask(__name__)
app.config["SWAGGER"] = {
"title": "Swagger Basic Auth App",
"uiversion": 2,
}
swag = Swagger(app,
decorators=[ requires_basic_auth ],
template={
"swagger": "2.0",
"info": {
"title": "Swagger Basic Auth App",
"version": "1.0",
},
"consumes": [
"application/json",
],
"produces": [
"application/json",
],
},
)
api_config = {
"headers": [
],
"specs": [
{
"endpoint": 'apispec',
"route": '/apispec.json',
"rule_filter": lambda rule: True, # all in
"model_filter": lambda tag: True, # all in
}
],
"static_url_path": "/flasgger_static",
"specs_route": "/" if config_name == 'api' else '/api/'
}
Swagger(app, config=api_config)
# # HTTP error handling
# @app.errorhandler(404)
# def not_found(error):
# return render_template('error/404.html'), 404
# @app.errorhandler(500)
# def internal_server_error(error):
# return render_template('error/500.html'), 500
# Import a module / component using its blueprint handler variable
from app.auth.controllers import mod as auth_module
from app.users.controllers import mod as user_module
# Register blueprint(s)
# app.register_blueprint(xyz_module)
def register_extensions(app):
"""Register Flask extensions"""
CORS(app)
db.init_app(app)
login_manager.init_app(app)
mail.init_app(app)
migrate.init_app(app, db)
pagedown.init_app(app)
swag = Swagger(app)
if os.environ.get('FLASK_CONFIG') != "production":
es = Elasticsearch([app.config['ES_URL']])
else:
es = Elasticsearch(
[app.config["ES_URL"]],
http_auth=(app.config["ES_USERNAME"], app.config["ES_PASSWORD"]),
)
app.elasticsearch = es
scrapyd = ScrapydAPI("http://localhost:6800")
scheduler = BackgroundScheduler()
app.scheduler = scheduler
app.scrapy = scrapyd
scheduler.start()
jwt.init_app(app)