python类Api()的实例源码

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

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)