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

How to use delete_file_share method in lisa

Best Python code snippet using lisa_python

file.py

Source:file.py

...248    return created_row249@router.delete(250    "/share/{share_uuid}",251    description="delete a file share")252async def delete_file_share(253        share_uuid: UUID,254        curr_user: models.User = Depends(get_current_active_user)):255    try:256        file_share = await crud.get_file_share_by_uuid(share_uuid)257        fake_path = (await file_share.fake_path.get()).path258        # makes sure user has access to path259        create_root_path(260            fake_path,261            get_settings().HOMES_PATH,262            get_settings().SHARED_PATH,263            curr_user.username,264        )265        await crud.delete_file_share(share_uuid)266    except (PathNotExists, DoesNotExist):267        raise HTTPException(268            status_code=status.HTTP_404_NOT_FOUND,269            detail="unknown file share uuid"270        ) from None271@router.get(272    "/share/{share_uuid}",273    response_model=schema.FileShare,274    description="get a file shares meta")275async def get_file_share_meta(276        share_uuid: UUID,277        curr_user: models.User = Depends(get_current_active_user)):278    try:279        file_share = await crud.get_file_share_by_uuid(share_uuid)280        fake_path = (await file_share.fake_path.get()).path281        # makes sure user has access to path282        create_root_path(283            fake_path,284            get_settings().HOMES_PATH,285            get_settings().SHARED_PATH,286            curr_user.username,287        )288        return file_share289    except (PathNotExists, DoesNotExist):290        raise HTTPException(291            status_code=status.HTTP_404_NOT_FOUND,292            detail="unknown file share uuid"293        ) from None294@router.get(295    "/share/{share_uuid}/download",296    response_class=FileResponse,297    description="get a file shares file")298async def get_file_share_file(share_uuid: UUID):299    try:300        file_share = await crud.get_file_share_by_uuid(share_uuid)301        fake_path = (await file_share.fake_path.get()).path302        # make sure share isn't expired or has run out of uses303        if ((file_share.expires is not None and file_share.expires < timezone.now()) or304                (file_share.uses_left is not None and file_share.uses_left < 1)):305            raise SharePathInvalid()306        full_path = create_root_path(307            fake_path,308            get_settings().HOMES_PATH,309            get_settings().SHARED_PATH,310        )311        if not full_path.exists():312            # remove the share from database as path no longer exists313            await crud.delete_file_share(share_uuid)314            raise PathNotExists()315        if file_share.uses_left is not None:316            # if the share has limited uses subtract one317            file_share.uses_left -= 1318            await file_share.save()319        return FileResponse(full_path, filename=full_path.name)320    except (PathNotExists, DoesNotExist):321        raise HTTPException(322            status_code=status.HTTP_404_NOT_FOUND,323            detail="unknown file share uuid"324        ) from None325    except SharePathInvalid:326        # share has either run out of uses or expired327        await crud.delete_file_share(share_uuid)328        raise HTTPException(329            status_code=status.HTTP_404_NOT_FOUND,330            detail="unknown file share uuid"...

storage_gateway.py

Source:storage_gateway.py

...66        elif len(shares) > 1:67            print(f"There are {len(shares)} NFS shares matched.")68        else:69            try:70                response = self.client.delete_file_share(FileShareARN=shares[0]['FileShareARN'],71                                                         ForceDelete=True)72            except Exception as e:73                self.logger.exception("")74                raise e75            else:76                print(f"NFS share has been deleted, S3 bucket name: {bucket_name}, Gateway: {self.config['name']}")77                return response78    def get_gateway_by_name(self) -> list:79        """80        Get gateway by gateway name81        :return: A list of gateways that match the gateway name82        """83        list_gateways = []84        try:...

cleaner.py

Source:cleaner.py

...22            blob_service.delete_container(container_name)23    except azure.core.exceptions.ResourceExistsError as err:24        print(err)25        pass26def delete_file_share(file_service, share_name):27    print("Delete share: {}/{}".format(file_service.account_name, share_name))28    file_service.delete_share(share_name)29def delete_queue(queue_service, queue_name):30    print("Delete queue: {}/{}".format(queue_service.account_name, queue_name))31    queue_service.delete_queue(queue_name)32def delete_table(table_service, table_name):33    print("Delete table: {}/{}".format(table_service.account_name, table_name))34    table_service.delete_table(table_name)35def service_exists(service_client):36    try:37        socket.getaddrinfo(urllib.parse.urlparse(38            service_client.url).netloc, 443)39    except socket.gaierror:40        return False41    return True42def clean_storage_account(connection_string):43    pool = ThreadPool(16)44    blob_service = BlobServiceClient.from_connection_string(connection_string)45    if service_exists(blob_service):46        pool.map(lambda container: delete_container(blob_service,47                 container.name), blob_service.list_containers())48    file_service = ShareServiceClient.from_connection_string(connection_string)49    if service_exists(file_service):50        pool.map(lambda share: delete_file_share(51            file_service, share.name), file_service.list_shares())52    queue_service = QueueServiceClient.from_connection_string(53        connection_string)54    if service_exists(queue_service):55        pool.map(lambda queue: delete_queue(queue_service,56                 queue.name), queue_service.list_queues())57    table_service = TableServiceClient.from_connection_string(58        connection_string)59    if service_exists(table_service):60        try:61            pool.map(lambda table: delete_table(table_service,62                     table.name), table_service.list_tables())63        except azure.core.exceptions.HttpResponseError as e:64            pass...

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.