Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_function_version method in localstack

Best Python code snippet using localstack_python

function_version.py

Source:function_version.py

...153        download = strutils.bool_from_string(154            pecan.request.GET.get('download', False)155        )156        version = int(version)157        version_db = db_api.get_function_version(function_id, version)158        if not download:159            LOG.info("Getting version %s for function %s.", version,160                     function_id)161            pecan.override_template('json')162            return resources.FunctionVersion.from_db_obj(version_db).to_dict()163        LOG.info("Downloading version %s for function %s.", version,164                 function_id)165        f = self.storage_provider.retrieve(version_db.project_id, function_id,166                                           None, version=version)167        if isinstance(f, collections.Iterable):168            pecan.response.app_iter = f169        else:170            pecan.response.app_iter = FileIter(f)171        pecan.response.headers['Content-Type'] = 'application/zip'172        pecan.response.headers['Content-Disposition'] = (173            'attachment; filename="%s_%s"' % (function_id, version)174        )175    @rest_utils.wrap_wsme_controller_exception176    @wsme_pecan.wsexpose(None, types.uuid, int, status_code=204)177    def delete(self, function_id, version):178        """Delete a specific function version.179        - The version should not being used by any job180        - The version should not being used by any webhook181        - Admin user can not delete normal user's version182        """183        ctx = context.get_ctx()184        acl.enforce('function_version:delete', ctx)185        LOG.info("Deleting version %s of function %s.", version, function_id)186        with db_api.transaction():187            version_db = db_api.get_function_version(function_id, version,188                                                     insecure=False)189            latest_version = version_db.function.latest_version190            version_jobs = db_api.get_jobs(191                function_id=version_db.function_id,192                function_version=version_db.version_number,193                status={'nin': ['done', 'cancelled']}194            )195            if len(version_jobs) > 0:196                raise exc.NotAllowedException(197                    'The function version is still associated with running '198                    'job(s).'199                )200            version_webhook = db_api.get_webhooks(201                function_id=version_db.function_id,202                function_version=version_db.version_number,203            )204            if len(version_webhook) > 0:205                raise exc.NotAllowedException(206                    'The function version is still associated with webhook.'207                )208            filters = rest_utils.get_filters(209                function_id=version_db.function_id,210                function_version=version_db.version_number211            )212            version_aliases = db_api.get_function_aliases(**filters)213            if len(version_aliases) > 0:214                raise exc.NotAllowedException(215                    'The function version is still associated with alias.'216                )217            # Delete resources for function version218            self.engine_client.delete_function(function_id, version=version)219            etcd_util.delete_function(function_id, version=version)220            self.storage_provider.delete(ctx.projectid, function_id, None,221                                         version=version)222            db_api.delete_function_version(function_id, version)223            if latest_version == version:224                version_db.function.latest_version = latest_version - 1225        LOG.info("Version %s of function %s deleted.", version, function_id)226    @rest_utils.wrap_wsme_controller_exception227    @wsme_pecan.wsexpose(228        None,229        types.uuid,230        int,231        body=resources.ScaleInfo,232        status_code=202233    )234    def scale_up(self, function_id, version, scale):235        """Scale up the workers for function version execution.236        This is admin only operation. The load monitoring of execution237        depends on the monitoring solution of underlying orchestrator.238        """239        acl.enforce('function_version:scale_up', context.get_ctx())240        func_db = db_api.get_function(function_id)241        # If version=0, it's equivalent to /functions/<funcion-id>/scale_up242        if version > 0:243            db_api.get_function_version(function_id, version)244        params = scale.to_dict()245        LOG.info('Starting to scale up function %s(version %s), params: %s',246                 function_id, version, params)247        self.engine_client.scaleup_function(248            function_id,249            runtime_id=func_db.runtime_id,250            version=version,251            count=params['count']252        )253    @rest_utils.wrap_wsme_controller_exception254    @wsme_pecan.wsexpose(255        None,256        types.uuid,257        int,258        body=resources.ScaleInfo,259        status_code=202260    )261    def scale_down(self, function_id, version, scale):262        """Scale down the workers for function version execution.263        This is admin only operation. The load monitoring of execution264        depends on the monitoring solution of underlying orchestrator.265        """266        acl.enforce('function_version:scale_down', context.get_ctx())267        db_api.get_function(function_id)268        params = scale.to_dict()269        # If version=0, it's equivalent to /functions/<funcion-id>/scale_down270        if version > 0:271            db_api.get_function_version(function_id, version)272        workers = etcd_util.get_workers(function_id, version=version)273        if len(workers) <= 1:274            LOG.info('No need to scale down function %s(version %s)',275                     function_id, version)276            return277        LOG.info('Starting to scale down function %s(version %s), params: %s',278                 function_id, version, params)279        self.engine_client.scaledown_function(function_id, version=version,280                                              count=params['count'])281    @rest_utils.wrap_wsme_controller_exception282    @wsme_pecan.wsexpose(None, types.uuid, int, status_code=202)283    def detach(self, function_id, version):284        """Detach the function version from its underlying workers.285        This is admin only operation, which gives admin user a safe way to286        clean up the underlying resources allocated for the function version.287        """288        acl.enforce('function_version:detach', context.get_ctx())289        db_api.get_function(function_id)290        # If version=0, it's equivalent to /functions/<funcion-id>/detach291        if version > 0:292            db_api.get_function_version(function_id, version)293        LOG.info('Starting to detach function %s(version %s)', function_id,294                 version)295        # Delete allocated resources in orchestrator and etcd keys.296        self.engine_client.delete_function(function_id, version=version)...

function.py

Source:function.py

...154    }155    function = conn.fgs.execute_function_asynchronously(function_urn,**function_req)156    return  function157# Querying the Aliases of a Function's All Versions158def get_function_version(function_urn):159    function_list = conn.fgs.get_function_version(function_urn=function_urn,marker=0,maxitems=400)160    for data in function_list:161        print(data)162    return None163if __name__ == "__main__":164    #Querying a Function List165    get_function_list()166    #Creating a Function167    func = create_function()168    if func.func_urn:169        #Querying the Metadata Information of a Function170        func_get_conf = get_function_metadata(func.func_urn)171        print("func_get_conf:")172        print(func_get_conf)173        #Executing a Function Synchronously174        execut_function_syn = execut_function_synchronously(func.func_urn)175        print("execut_function_synchronously:")176        print(execut_function_syn)177        #Modifying the Code of a Function178        func_update_conf = update_function_conf(func.func_urn)179        print("func_update_conf:")180        print(func_update_conf)181        #Publishing a Function Version182        func_public_version = public_function_verion(func.func_urn)183        print("public_function_verion:")184        print(func_public_version)185        #Querying the Aliases of a Function's All Versions186        get_function_version=get_function_version(func.func_urn)187        print("get_function_version:")188        print(get_function_version)189        #Querying the Code of a Function190        func_get_code = get_function_code(func.func_urn)191        print("func_get_code:")192        print(func_get_code)193        #Modifying the Code of a Function194        func_update_code = update_function_code(func.func_urn)195        print("func_update_code:")196        print(func_update_code)197        #Creating an Alias for a Function Version198        func_create_aliase = create_function_aliase(func.func_urn)199        print("create_function_aliase:")200        print(func_create_aliase)...

__init__.py

Source:__init__.py

1"""Subpackage for function and layer service status and management functionality."""2from .functioning import echo_function_versions  # noqa: F4013from .functioning import get_function_version  # noqa: F4014from .functioning import get_function_versions  # noqa: F4015from .functioning import remove_function_version  # noqa: F4016from .layering import echo_layer_versions  # noqa: F4017from .layering import get_layer_version  # noqa: F4018from .layering import get_layer_versions  # noqa: F401...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.