How to use delete_partitions method in lisa

Best Python code snippet using lisa_python

arg_parser.py

Source:arg_parser.py

1import abc2import argparse3from abc import ABC4class Action(ABC):5    @abc.abstractmethod6    def run(self, args: argparse.Namespace):7        pass8class ArgumentParser:9    def __init__(10        self,11        switch_action: Action,12        display_action: Action,13        delete_partitions_action: Action,14        copy_partitions: Action,15    ):16        self.switch_action = switch_action17        self.display_action = display_action18        self.delete_partitions_action = delete_partitions_action19        self.copy_partitions_action = copy_partitions20    def build_parser(self, args=None) -> argparse.Namespace:21        parser = argparse.ArgumentParser(22            description="Helper to automate some common bigquery tasks"23        )24        subparsers = parser.add_subparsers()25        display = subparsers.add_parser("display")26        display.add_argument("--project-id", required=True)27        display.add_argument("--dataset-id", required=True)28        display.add_argument("--view", required=True)29        display.set_defaults(func=self.display_action.run)30        switch = subparsers.add_parser("switch-color")31        switch.add_argument("--project-id", required=True)32        switch.add_argument("--dataset-id", required=True)33        switch.add_argument("--view", required=True)34        switch.add_argument("--rebuild", action="store_true")35        switch.set_defaults(func=self.switch_action.run)36        delete_partitions = subparsers.add_parser("delete-partitions")37        delete_partitions.add_argument("--project-id", required=True)38        delete_partitions.add_argument("--dataset-id", required=True)39        delete_partitions.add_argument("--table", required=True)40        delete_partitions.add_argument("--start-date", required=True)41        delete_partitions.add_argument("--end-date", required=True)42        delete_partitions.set_defaults(func=self.delete_partitions_action.run)43        copy_partitions = subparsers.add_parser("copy-partitions")44        copy_partitions.add_argument("--src-project-id", required=True)45        copy_partitions.add_argument("--dst-project-id", required=True)46        copy_partitions.add_argument("--dataset-id", required=True)47        copy_partitions.add_argument("--table", required=True)48        copy_partitions.add_argument("--start-date", required=True)49        copy_partitions.add_argument("--end-date", required=True)50        copy_partitions.set_defaults(func=self.copy_partitions_action.run)...

utils.py

Source:utils.py

1import google.cloud.storage2import requests3import uuid4import settings5class PartitionedBlob:6    COMPOSE_LIMIT = 327    def __init__(self, bucket_name, directory="tmp", respect_compose_limit=True):8        self.bucket_name = bucket_name9        self.directory = directory10        self.respect_compose_limit = respect_compose_limit11        client = google.cloud.storage.Client()12        self.bucket = client.get_bucket(bucket_name)13        self.blobs = []14    def append_blob(self, blob):15        if len(self.blobs) >= self.COMPOSE_LIMIT:16            self.compose(self._make_tmp_blob_name())17        self.blobs.append(blob)18        return blob19    def append_contents(self, contents, **upload_kwargs):20        blob = self._make_blob()21        blob.upload_from_string(contents, **upload_kwargs)22        return self.append_blob(blob)23    def append_file(self, path, **upload_kwargs):24        blob = self._make_blob()25        blob.upload_from_filename(path, **upload_kwargs)26        return self.append_blob(blob)27    def compose(self, path, delete_partitions=True):28        blob = self.bucket.blob(path)29        blob.compose(self.blobs)30        if delete_partitions:31            for _blob in self.blobs:32                _blob.delete()33        self.blobs = [blob]34        return blob35    def _make_blob(self, path=None):36        if path is None:37            path = self._make_tmp_blob_name()38        blob = self.bucket.blob(path)39        return blob40    def _make_tmp_blob_name(self):41        blob_name = uuid.uuid4()42        return f"""{self.directory}/{blob_name}"""43def stream_upload(source_url, destination_path, tmp_path="tmp", bucket_name=None,44                  chunk_size=None):45    if bucket_name is None:46        bucket_name = settings.PODCAST_STORAGE_BUCKET47    if chunk_size is None:48        chunk_size = settings.STREAM_UPLOAD_CHUNK_SIZE49    pblob = PartitionedBlob(bucket_name=bucket_name, directory=tmp_path, respect_compose_limit=True)50    request = requests.get(source_url, stream=True)51    content_type = request.headers["Content-Type"]52    stream = request.iter_content(chunk_size=chunk_size)53    for chunk in stream:54        pblob.append_contents(chunk)55    blob = pblob.compose(destination_path, delete_partitions=True)56    blob.content_type = content_type57    blob.update()...

stop_replica_request.py

Source:stop_replica_request.py

1from typing import ClassVar, List2from ...constants import ApiKey3from ..base import RequestData4class Partition:5    topic: str6    partition_ids: List[int]7    def __init__(self, topic: str, partition_ids: List[int]):8        """9        :param topic: Name of topic10        :type topic: str11        :param partition_ids: The partition ids of a topic12        :type partition_ids: List[int]13        """14        self.topic = topic15        self.partition_ids = partition_ids16class StopReplicaRequestData(RequestData):17    controller_id: int18    controller_epoch: int19    broker_epoch: int20    delete_partitions: bool21    partitions: List[Partition]22    api_key: ClassVar[ApiKey] = ApiKey.STOP_REPLICA23    def __init__(24        self,25        controller_id: int,26        controller_epoch: int,27        broker_epoch: int,28        delete_partitions: bool,29        partitions: List[Partition],30    ):31        """32        :param controller_id: The controller id33        :type controller_id: int34        :param controller_epoch: The controller epoch35        :type controller_epoch: int36        :param broker_epoch: The broker epoch37        :type broker_epoch: int38        :param delete_partitions: Boolean which indicates if replica's partitions must be deleted.39        :type delete_partitions: bool40        :param partitions: The partitions41        :type partitions: List[Partition]42        """43        self.controller_id = controller_id44        self.controller_epoch = controller_epoch45        self.broker_epoch = broker_epoch46        self.delete_partitions = delete_partitions...

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.