KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_storage_client method in lisa

Best Python code snippet using lisa_python

completers.py

Source:completers.py

...11    if not namespace.share_name:12        return []13    validate_client_parameters(cmd, namespace)14    t_file_service = cmd.get_models('file#FileService')15    client = get_storage_client(cmd.cli_ctx, t_file_service, namespace)16    share_name = namespace.share_name17    directory_name = prefix or ''18    try:19        items = list(client.list_directories_and_files(share_name, directory_name))20    except AzureMissingResourceHttpError:21        directory_name = directory_name.rsplit('/', 1)[0] if '/' in directory_name else ''22        items = list(client.list_directories_and_files(share_name, directory_name))23    path_format = '{}{}' if directory_name.endswith('/') or not directory_name else '{}/{}'24    names = []25    for i in items:26        name = path_format.format(directory_name, i.name)27        if not hasattr(i.properties, 'content_length'):28            name = '{}/'.format(name)29        names.append(name)30    return sorted(names)31@Completer32def dir_path_completer(cmd, prefix, namespace):33    from azure.common import AzureMissingResourceHttpError34    if not namespace.share_name:35        return []36    validate_client_parameters(cmd, namespace)37    t_file_service = cmd.get_models('file#FileService')38    client = get_storage_client(cmd.cli_ctx, t_file_service, namespace)39    share_name = namespace.share_name40    directory_name = prefix or ''41    try:42        items = list(client.list_directories_and_files(share_name, directory_name))43    except AzureMissingResourceHttpError:44        directory_name = directory_name.rsplit('/', 1)[0] if '/' in directory_name else ''45        items = list(client.list_directories_and_files(share_name, directory_name))46    dir_list = [x for x in items if not hasattr(x.properties, 'content_length')]47    path_format = '{}{}/' if directory_name.endswith('/') or not directory_name else '{}/{}/'48    names = []49    for d in dir_list:50        name = path_format.format(directory_name, d.name)51        names.append(name)52    return sorted(names)53def get_storage_name_completion_list(service, func, parent=None):54    @Completer55    def completer(cmd, _, namespace):56        validate_client_parameters(cmd, namespace)57        client = get_storage_client(cmd.cli_ctx, service, namespace)58        if parent:59            parent_name = getattr(namespace, parent)60            method = getattr(client, func)61            items = [x.name for x in method(**{parent: parent_name})]62        else:63            items = [x.name for x in getattr(client, func)()]64        return items65    return completer66def get_storage_acl_name_completion_list(service, container_param, func):67    @Completer68    def completer(cmd, _, namespace):69        validate_client_parameters(cmd, namespace)70        client = get_storage_client(cmd.cli_ctx, service, namespace)71        container_name = getattr(namespace, container_param)72        return list(getattr(client, func)(container_name))...

gcs_util.py

Source:gcs_util.py

...3from google.cloud.storage import Client4from google.cloud.storage.bucket import Bucket5from google.cloud.storage.blob import Blob6LOGGER = getLogger('airflow.task')7def get_storage_client(project_id='och-sandbox'):8    return Client(project=project_id)9def create_bucket(bucket_name, project_id='och-sandbox'):10    """11    create a Storage Bucket12    :param bucket_name: name of the bucket to create13    :param project_id: default is och-sandbox14    :return: Bucket Object15    """16    client = get_storage_client(project_id)17    bucket = client.create_bucket(bucket_name, location='EU')18    LOGGER.info(f"{project_id}.{bucket_name} has been created !")19    return bucket20def list_buckets(project_id='och-sandbox'):21    """22    return the list of the IDs of the buckets of the project23    :param project_id: default is och-sandbox24    :return: List of string representing bucket Ids25    """26    client = get_storage_client(project_id)27    bucket_iterator = client.list_buckets()28    ls_bucket_id = [e.id for e in bucket_iterator]29    return ls_bucket_id30def get_blob(bucket_name, blob_name, project_id='och-sandbox'):31    """32    get a blob in a bucket33    :param bucket_name:34    :param blob_name:35    :param project_id: default is och-sandbox36    :return:37    """38    client_storage = get_storage_client(project_id)39    bucket = client_storage.get_bucket(bucket_name)40    return bucket.get_blob(blob_name)41def list_blobs(bucket_name, prefix=None, project_id='och-sandbox') :42    """43    list blobs in a bucket, with a prefix.44    all files matching the prefix will be returned45    :param bucket_name: Name of the bucket46    :param prefix: Blobs prefix47    :param project_id: defaulting to och-sandbox48    :return: List containing the blobs49    """50    storage_client = get_storage_client(project_id)51    LOGGER.info(f"Listing blobs from bucket {project_id} {bucket_name} {prefix}")52    ls = list(storage_client.list_blobs(bucket_or_name=bucket_name, prefix=prefix))53    return ls54def upload_blob_from_filename(bucket_name, source_file_name, destination_blob_name, project_id='och-sandbox') :55    """56    upload a local file to a GCS bucket57    :param bucket_name:58    :param source_file_name: local file to upload59    :param destination_blob_name: full path of the blob including the name of the file60    :param project_id defaulting to och-sandbox61    :return:62    """63    storage_client = get_storage_client(project_id=project_id)64    bucket = storage_client.get_bucket(bucket_or_name=bucket_name)65    blob = bucket.blob(destination_blob_name)66    blob.upload_from_filename(source_file_name)...

resources.py

Source:resources.py

...8api = Api(app=app)9class Articles(Resource):10    @marshal_with(responses.Article.resource_fields)11    def get(self):12        return get_storage_client().get_articles(), 20013class Article(Resource):14    @marshal_with(responses.Article.resource_fields)15    def get(self, title=None):16        return get_storage_client().get_article(title=title)17    @marshal_with(responses.Article.resource_fields)18    def delete(self, title=None):19        return get_storage_client().delete_article(title=title), 20220    @marshal_with(responses.Article.resource_fields)21    def put(self, title=None):22        form = request.form23        content = form.get('content')24        return get_storage_client().create_article(title=title,25                                                   content=content), 20126    @marshal_with(responses.Article.resource_fields)27    def post(self, title=None):28        data = request.form29        content = data.get('content')30        new_title = data.get('title')31        return get_storage_client().update_article(title,32                                                   new_title=new_title,33                                                   content=content), 20034api.add_resource(Article, '/article/<title>', endpoint='article/<title>')35api.add_resource(Articles, '/articles', endpoint='articles')36def handle_invalid_usage(error):37    response = error.__dict__38    return json.dumps(response), response.get('status_code', 500)39api.handle_error = handle_invalid_usage40if __name__ == '__main__':41    init_db()...

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.