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

How to use container_client method in localstack

Best Python code snippet using localstack_python

createPopContainersAzure.py

Source:createPopContainersAzure.py

1import sys, os, time2import config as cfg3from azure.storage.blob import BlobServiceClient, ContainerClient4def create_container(BlobServiceClient, container_name):5    container_client = blob_service_client.get_container_client(container_name)6    try:7        container_properties = container_client.get_container_properties()8    except Exception as e:9        container_client = blob_service_client.create_container(container_name)10    return container_client11def upload_all_blobs_from_folder(container_client, folder_path):12    blob_list = [blob.name for blob in container_client.list_blobs()]13    for filename in os.listdir(folder_path):14        if filename not in blob_list:15            upload_file_path = os.path.join(folder_path, filename)16            upload_file(container_client, upload_file_path, filename)17def upload_file(container_client, file_path, object_name):18    try:19        with open(file_path, "rb") as file:20            container_client.upload_blob(name=object_name, data=file)21    except Exception as e:22        print(e)23def teardown(blob_service_client):24    for item in cfg.task_one['containers']:25        container_name = item['container']['name']26        container_client = blob_service_client.get_container_client(container_name)27        for blob in container_client.list_blobs():28            container_client.delete_blob(blob.name)29            print("Blob: '" + blob.name + "' successfully deleted.")30        container_client.delete_container()31        print("\tContainer: '" + container_name + "' successfully deleted.")32if __name__ == "__main__":33    try:34        connect_str = os.getenv('AZURE_STORAGE_CONNECTION_STRING')35        blob_service_client = BlobServiceClient.from_connection_string(connect_str)36        37        if len(sys.argv) > 1 and sys.argv[1] == "reset":38            start_time = time.time()39            teardown(blob_service_client)40            print(str(time.time() - start_time) + " seconds.")...

db_util.py

Source:db_util.py

...6HOST = cosmos_config.settings['host']7MASTER_KEY = cosmos_config.settings['master_key']8DATABASE_ID = cosmos_config.settings['database_id']9CONTAINER_ID = cosmos_config.settings['container_id']10def create_database_container_client():11    client = cosmos_client.CosmosClient(HOST, {'masterKey': MASTER_KEY}, user_agent="CosmosDBDotnetQuickstart",12                                        user_agent_overwrite=True)13    db = client.get_database_client(DATABASE_ID)14    return db.get_container_client(CONTAINER_ID)15def recreate_database_data(category_with_books):16    container_client = create_database_container_client()17    logging.info("Clearing database")18    clear_database(container_client)19    for category_and_book in category_with_books:20        logging.info(f"Saving category: {category_and_book[0].technical_name}")21        save_category(category_and_book[0], container_client)22        for book in category_and_book[1]:23            logging.info(f"Saving book '{book.technical_name} to category '{category_and_book[0].technical_name}'")24            save_book(book, container_client)25def clear_database(container_client):26    documents = list(container_client.query_items(query="SELECT * FROM c", enable_cross_partition_query=True))27    for document in documents:28        container_client.delete_item(29            item=document['id'], partition_key=document['category_num'])30def save_category(category_data: Category, container_client):...

category_service.py

Source:category_service.py

1from util import db_util2def get_categories():3    container_client = db_util.create_database_container_client()4    return container_client.query_items(query="SELECT * FROM c WHERE c._type='Category'",5                                        enable_cross_partition_query=True)6def get_categories_by_id(category_id):7    container_client = db_util.create_database_container_client()8    return container_client.query_items(query="SELECT * FROM c WHERE c._type='Category' AND c.id=@categoryId",9                                        enable_cross_partition_query=True, parameters=[10            {"name": "@categoryId", "value": category_id}11        ])12def get_categories_by_category_num(category_num):13    container_client = db_util.create_database_container_client()14    return container_client.query_items(15        query="SELECT * FROM c WHERE c._type='Category' AND c.category_num=@categoryNum",16        parameters=[17            {"name": "@categoryNum", "value": category_num}18        ])19def get_category_basic_data_by_category_num():20    container_client = db_util.create_database_container_client()21    return container_client.query_items(22        query="SELECT c.id, c.category_num, c.technical_name FROM c WHERE c._type='Category'",...

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.