def build_app(configure_logging=True, with_keystone=True):
app = flask.Flask(__name__)
app.url_map.converters.update(converters.ALL)
api.init_app(app) # init_app spoils Api object if app is a blueprint
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False # silence warning
# TUNINGBOX_SETTINGS is the path to the file with tuning_box configuration
app.config.from_envvar('TUNINGBOX_SETTINGS', silent=True)
# These handlers work if PROPAGATE_EXCEPTIONS is on (Nailgun case)
app.register_error_handler(sa_exc.IntegrityError, handle_integrity_error)
app.register_error_handler(errors.TuningboxIntegrityError,
handle_integrity_error)
app.register_error_handler(errors.TuningboxNotFound,
handle_object_not_found)
app.register_error_handler(errors.RequestValidationError,
handle_request_validation_error)
app.register_error_handler(errors.KeysOperationError,
handle_keys_operation_error)
db.db.init_app(app)
if configure_logging:
log_level = app.config.get('LOG_LEVEL', 'DEBUG')
logger.init_logger(app, log_level)
if with_keystone:
app.wsgi_app = keystone.KeystoneMiddleware(app)
return app
python类Api()的实例源码
def deferred(self):
"""Runs the API, listens to external requests."""
app = Flask('argosd')
api = flask_restful.Api(app)
api.add_resource(ShowsResource, '/shows')
api.add_resource(ShowResource, '/shows/<int:show_id>')
api.add_resource(EpisodesResource, '/episodes')
logfile = '{}/api.log'.format(settings.LOG_PATH)
logformat = '%(message)s'
logging.basicConfig(format=logformat, level=logging.INFO,
filename=logfile, filemode='a')
app.run(host='0.0.0.0', port=27467)
def __init__(self, debug, core, host, port, app_root):
self.core = core
self.flask = Flask("PyPush.web",
static_folder=os.path.join(const.PUSH_WEB_DIR, "static"),
template_folder=os.path.join(const.PUSH_WEB_DIR, "templates"),
)
self.bower = Bower(self.flask)
self.restful = Api(self.flask)
self._applyDefaultConfig()
self.host = host
self.port = port
self.debug = debug
self.flask.config.update(
APPLICATION_ROOT=app_root,
)
def register_blueprints(app):
api_bp = Blueprint('api', __name__)
api = Api(api_bp)
api.add_resource(UserView, '/users', '/users/<int:id>')
api.add_resource(OrderView, '/orders', '/orders/<int:id>')
api.add_resource(ProductView, '/products', '/products/<int:id>')
api.add_resource(LoginView, '/login')
api.add_resource(LogoutView, '/logout')
app.register_blueprint(api_bp, subdomain='api')
app.register_blueprint(admin_security, subdomain='admin')
app.register_blueprint(
admin_static, subdomain='admin', url_prefix='/static')
app.register_blueprint(admin_user, subdomain='admin', url_prefix='/users')
app.register_blueprint(
admin_order, subdomain='admin', url_prefix='/orders')
app.register_blueprint(
admin_product, subdomain='admin', url_prefix='/products')
app.register_blueprint(
admin_category, subdomain='admin', url_prefix='/categories')
def __init__(self):
self.__app = Flask(__name__)
self.__api = Api(self.__app)
self.__parser = reqparse.RequestParser()
self.__stub_to_peer_service = None
# SSL ?? ??? ?? context ?? ??? ????.
if conf.ENABLE_REST_SSL == 0:
self.__ssl_context = None
elif conf.ENABLE_REST_SSL == 1:
self.__ssl_context = (conf.DEFAULT_SSL_CERT_PATH, conf.DEFAULT_SSL_KEY_PATH)
else:
self.__ssl_context = ssl.SSLContext(_ssl.PROTOCOL_SSLv23)
self.__ssl_context.verify_mode = ssl.CERT_REQUIRED
self.__ssl_context.check_hostname = False
self.__ssl_context.load_verify_locations(cafile=conf.DEFAULT_SSL_TRUST_CERT_PATH)
self.__ssl_context.load_cert_chain(conf.DEFAULT_SSL_CERT_PATH, conf.DEFAULT_SSL_KEY_PATH)
def ApiApp(trustlines):
app = Flask(__name__)
api_bp = Blueprint('api', __name__, url_prefix='/api/v1')
api = Api(api_bp)
api.add_resource(NetworkList, '/networks', resource_class_args=[trustlines])
api.add_resource(Network, '/networks/<address:address>', resource_class_args=[trustlines])
api.add_resource(UserList, '/networks/<address:address>/users', resource_class_args=[trustlines])
api.add_resource(User, '/networks/<address:network_address>/users/<address:user_address>', resource_class_args=[trustlines])
api.add_resource(ContactList, '/networks/<address:network_address>/users/<address:user_address>/contacts', resource_class_args=[trustlines])
api.add_resource(TrustlineList, '/networks/<address:network_address>/users/<address:user_address>/trustlines', resource_class_args=[trustlines])
api.add_resource(Trustline, '/networks/<address:network_address>/users/<address:a_address>/trustlines/<address:b_address>', resource_class_args=[trustlines])
api.add_resource(Spendable, '/networks/<address:network_address>/users/<address:a_address>/spendables', resource_class_args=[trustlines])
api.add_resource(SpendableTo, '/networks/<address:network_address>/users/<address:a_address>/spendables/<address:b_address>', resource_class_args=[trustlines])
api.add_resource(TransactionInfos, '/txinfos/<address:address>', resource_class_args=[trustlines])
api.add_resource(Block, '/blocknumber', resource_class_args=[trustlines])
api.add_resource(Relay, '/relay', resource_class_args=[trustlines])
api.add_resource(Balance, '/balance/<address:address>', resource_class_args=[trustlines])
app.url_map.converters['address'] = AddressConverter
app.register_blueprint(api_bp)
return app
def configure_api_from_blueprint(blueprint, route_tuples):
"""
Creates a Flask Restful api object based on information from given
blueprint. API is configured to return JSON objects.
Each blueprint is describe by a list of tuple. Each tuple is composed of a
route and the related resource (controller).
"""
api = Api(blueprint, catch_all_404s=True)
api.representations = {
'application/json; charset=utf-8': output_json,
'application/json': output_json,
}
for route_tuple in route_tuples:
(path, resource) = route_tuple
api.add_resource(resource, path)
return api
def post(self):
args = imovel_parser.parse_args()
if args['caracteristicas'] == [None]:
args['caracteristicas'] = None
imovel_id = len(IMOVEIS) + 1
imovel_id = str(imovel_id)
IMOVEIS[imovel_id] = {
'proprietario': args['proprietario'],
'caracteristicas': args['caracteristicas'],
'endereco': args['endereco'],
'valor': args['valor'],
'esta_ocupado': args['esta_ocupado'],
}
return IMOVEIS[imovel_id], 201
##
## Actually setup the Api resource routing here
##
def app_factory(API_NAME="api"):
"""Create an app object."""
app = Flask(__name__)
CORS(app)
app.url_map.strict_slashes = False
api = Api(app)
api.add_resource(Index, "/"+API_NAME+"/", endpoint="api")
api.add_resource(Vocab, "/"+API_NAME+"/vocab", endpoint="vocab")
api.add_resource(Contexts, "/"+API_NAME+"/contexts/<string:category>.jsonld", endpoint="contexts")
api.add_resource(Entrypoint, "/"+API_NAME+"/contexts/EntryPoint.jsonld", endpoint="main_entrypoint")
api.add_resource(ItemCollection, "/"+API_NAME+"/<string:type_>", endpoint="item_collection")
api.add_resource(Item, "/"+API_NAME+"/<string:type_>/<int:id_>", endpoint="item")
return app
def __init__(self, host, port):
self.logger = Logger.Logger('executor')
self.loadConfigs()
self.host = host
self.port = port
self.service = Flask('Executor')
self.service.config['BUNDLE_ERRORS'] = True
CORS(self.service)
got_request_exception.connect(logException, self.service)
self.api = Api(self.service)
self.api.add_resource(Leaves.LeafIndex, '/leaves/')
self.api.add_resource(Leaves.LeafRegister, '/leaves/register')
self.api.add_resource(Leaves.LeafResource, '/leaves/<int:leafId>')
self.api.add_resource(Projects.ProjectIndex, '/projects/')
self.api.add_resource(Projects.ProjectResource, '/projects/<int:projectId>')
self.api.add_resource(Builds.BuildIndex, '/builds/')
self.api.add_resource(Builds.BuildRecentIndex, '/builds/recent/')
self.api.add_resource(Builds.BuildResource, '/builds/<int:buildId>')
self.api.add_resource(CI.CI, '/ci/<int:projectId>')
def __init__(self, configRoot='.'):
self.__configDir = configRoot+'/.leafd/'
self.__configFname = self.__configDir+'config.yaml'
self.logger = Logger.Logger('LeafService')
if not os.path.exists(self.__configDir):
os.makedirs(self.__configDir)
self.__configure()
self.service = Flask('Leaf')
self.api = Api(self.service)
got_request_exception.connect(logException, self.service)
self.api.add_resource(Info.Config, '/config')
self.api.add_resource(Info.Status, '/status')
self.api.add_resource(Jobs.JobIndex, '/jobs/')
self.api.add_resource(Jobs.JobResource, '/jobs/<int:jobId>')
t = threading.Thread(name='Leaf service', target=appThread,
args=(self.service,Info.configStorage['port']))
t.start()
time.sleep(1)
self.__register()
t.join()
def __init__(self, config):
self.config = config
self.clouds = []
self.app = Flask(__name__)
self.api = swagger.docs(Api(self.app),
apiVersion='0.2.0')
self.resources = {}
def api_add_resource():
"""
The resource has multiple URLs and you can pass multiple URLs to the
add_resource() method on the Api object. Each one will be routed to
your Resource
"""
for url_pattern in URLPATTERNS:
try:
API.add_resource(
get_resource(url_pattern.target), url_pattern.url,
endpoint=get_endpoint(url_pattern.url))
except StopIteration:
LOGGER.error('url resource not found: %s', url_pattern.url)
def post(self):
args = parser.parse_args()
todo_id = int(max(TODOS.keys()).lstrip('todo')) + 1
todo_id = 'todo%i' % todo_id
TODOS[todo_id] = {'task': args['task']}
return TODOS[todo_id], 201
##
## Actually setup the Api resource routing here
##
def m_flask_restful(mocker):
return mocker.MagicMock(spec=FlaskRestful)
def _create_app(volume_manager, testing=False):
"""Factory method to create the Flask app and register all dependencies.
Args:
volume_manager (VolumeManager): The volume manager to be used withing the API controller
testing (bool): Whether or not to set the `TESTING` flag on the newly generated Flask application
Returns:
Flask: The application with all the needed dependencies bootstrapped
"""
unhandled_exception_errors = {
'EtcdConnectionFailed': {
'message': "The ETCD cluster is not responding.",
'status': 503,
}
}
config = {
'RESTFUL_JSON': {
'separators': (', ', ': '),
'indent': 2,
'sort_keys': False
},
'TESTING': testing
}
app = Flask(__name__)
app.config.update(**config)
api = RestApi(app, errors=unhandled_exception_errors, catch_all_404s=True)
Api._register_resources(api)
app.volume_manager = volume_manager
app.api = api
return app
def make_api(app, url_prefix='/'):
api = Api(app)
api.representation('application/json')(json_representation)
api.add_resource(HoudiniNodeshapeConverter, url_prefix)
return api
def get(self):
rdic = {}
for item in tmodel.get_todo_list():
rdic[item[0]] = {
"id": item[0],
"Name": item[1],
"Content": item[2]
}
return jsonify(rdic)
##
## Actually setup the Api resource routing here
##
def start(host, port, debug):
logger.info('Start {host}:{port}'.format(host=host, port=port))
api = Api(app)
api.add_resource(AddJob, '/api/add')
api.add_resource(JobStatus, '/api/status')
api.add_resource(FileUpload, '/api/upload')
api.add_resource(ResultData, '/api/list')
api.add_resource(ResultDetail, '/api/detail')
api.add_resource(Search, '/api/search')
# consumer
threads = []
for i in range(5):
threads.append(threading.Thread(target=consumer, args=()))
for i in threads:
i.setDaemon(daemonic=True)
i.start()
try:
global running_port, running_host
running_host = host if host != '0.0.0.0' else '127.0.0.1'
running_port = port
app.run(debug=debug, host=host, port=int(port), threaded=True, processes=1)
except socket.error as v:
if v.errno == errno.EACCES:
logger.critical('[{err}] must root permission for start API Server!'.format(err=v.strerror))
exit()
else:
logger.critical('{msg}'.format(msg=v.strerror))
logger.info('API Server start success')
def __init__(self, inc_ip, inc_port, manage):
# setup Flask
self.app = Flask(__name__)
self.api = Api(self.app)
self.ip = inc_ip
self.port = inc_port
self.manage = manage
self.playbook_file = '/tmp/son-emu-requests.log'
self.api.add_resource(ChainVersionsList, "/",
resource_class_kwargs={'api': self})
self.api.add_resource(ChainList, "/v1/chain/list",
resource_class_kwargs={'api': self})
self.api.add_resource(ChainVnfInterfaces, "/v1/chain/<src_vnf>/<src_intfs>/<dst_vnf>/<dst_intfs>",
resource_class_kwargs={'api': self})
self.api.add_resource(ChainVnfDcStackInterfaces,
"/v1/chain/<src_dc>/<src_stack>/<src_vnf>/<src_intfs>/<dst_dc>/<dst_stack>/<dst_vnf>/<dst_intfs>",
resource_class_kwargs={'api': self})
self.api.add_resource(BalanceHostList, "/v1/lb/list",
resource_class_kwargs={'api': self})
self.api.add_resource(BalanceHost, "/v1/lb/<vnf_src_name>/<vnf_src_interface>",
resource_class_kwargs={'api': self})
self.api.add_resource(BalanceHostDcStack, "/v1/lb/<src_dc>/<src_stack>/<vnf_src_name>/<vnf_src_interface>",
resource_class_kwargs={'api': self})
self.api.add_resource(QueryTopology, "/v1/topo",
resource_class_kwargs={'api': self})
self.api.add_resource(Shutdown, "/shutdown")
@self.app.after_request
def add_access_control_header(response):
response.headers['Access-Control-Allow-Origin'] = '*'
return response
def __init__(self, listenip, port):
self.ip = listenip
self.port = port
self.compute = None
self.manage = None
self.playbook_file = '/tmp/son-emu-requests.log'
with open(self.playbook_file, 'w'):
pass
# setup Flask
self.app = Flask(__name__)
self.api = Api(self.app)
def __init__(self):
self.__app = Flask(__name__)
self.__api = Api(self.__app)
self.__parser = reqparse.RequestParser()
self.__stub_to_rs_service = None
# SSL ?? ??? ?? context ?? ??? ????.
if conf.ENABLE_REST_SSL is True:
self.__ssl_context = (conf.DEFAULT_SSL_CERT_PATH, conf.DEFAULT_SSL_KEY_PATH)
else:
self.__ssl_context = None
def create_app(self):
"""This method creates the flask app.
This is required to be implemented by flask_restful.
:returns: Flask application instance.
"""
app = Flask('test')
app.config['TESTING'] = True
api = flask_restful.Api(app)
for url, handler in self.get_routes():
api.add_resource(handler, url)
return api.app
def create_app(object_name):
"""
An flask application factory
object_name: the python path of the config object,
e.g. oniongate.settings.ProdConfig
"""
app = Flask(__name__, instance_relative_config=True)
app.config.from_object(object_name)
try:
app.config.from_pyfile("config.py")
except FileNotFoundError:
pass
# Create zone file directory if it doesn't exist
zone_directory = app.config.get('zone_dir') or os.path.join(app.instance_path, 'zones')
if not os.path.isdir(zone_directory):
os.makedirs(zone_directory)
app.config["zone_dir"] = zone_directory
api_bp = Blueprint('api', __name__)
api = Api(api_bp)
CORS(app, resources={r"/api/*": {"origins": "*"}})
db.init_app(app)
# register our blueprints
app.register_blueprint(main_bp)
app.register_blueprint(api_bp, url_prefix='/api/v1')
api.add_resource(Domains, '/domains', '/domains/<domain_name>')
api.add_resource(Records, '/records/<domain_name>', '/records/<domain_name>/<record_id>')
api.add_resource(Proxies, '/proxies', '/proxies/<ip_address>')
app.jinja_env.filters['naturaltime'] = humanize.naturaltime
return app
def create_app(is_hd=True):
state.set_model(is_hd)
app = Flask(__name__)
app.register_blueprint(index)
app.config['ERROR_404_HELP'] = False
api = Api(app)
api.add_resource(SetProgram, '/api/program/<string:program>')
api.add_resource(StopProgram, '/api/program')
atexit.register(state.stop_program)
return app
def create_app(config):
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = get_meta_db_config_path(config)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app_config = get_application_config(config)
for key, value in app_config:
app.config[key.upper()] = int(value) if key.upper() in INT_OPTIONS else value
app.config['SENTRY_INCLUDE_PATHS'] = [
'postgraas_server',
]
app.config['SENTRY_RELEASE'] = postgraas_server.__version__
sentry.init_app(app)
from raven.handlers.logging import SentryHandler
app.logger.addHandler(SentryHandler(client=sentry.client, level=logging.WARN))
restful_api = Api(app)
restful_api.add_resource(DBInstanceResource, "/api/v2/postgraas_instances/<int:id>")
restful_api.add_resource(DBInstanceCollectionResource, "/api/v2/postgraas_instances")
db.init_app(app)
app.postgraas_backend = get_backend(config)
@app.route('/health')
def health():
return "ok"
return app
def parse_api():
parser = reqparse.RequestParser()
parser.add_argument('name', required=True, type=str,
help="Api needs name!",
location="json")
parser.add_argument('file_content', required=True, type=str,
help="must not be empty api!",
location="json")
return parser.parse_args()
def __init__(self):
loginfo("Creating RestApi instance.")
self.app = Flask("aide", static_url_path="")
self.api = Api(self.app, catch_all_404s=True)
CORS(self.app)
build_resources(self.api)
def create_app():
"""Creates the Flask app object."""
app = Flask(__name__)
app.config.from_object(AppConfig)
api = Api(app)
configure_resources(api)
configure_extensions(app)
return app
def configure(self, payload):
if self.level == 'app' or self.level == 'api':
self.server = Flask(__name__)
if self.level == 'app':
for api in self.app.get_apis():
setattr(api, 'server', Blueprint(api.name, __name__))
resource = Api(api.server)
for route in api.config.routes:
handler = api.get_handler(api, route.handler)
resource.add_resource(handler, api.config.main.prefix + route.path)
self.server.register_blueprint(api.server)