Module pachyderm_sdk.api.transaction

# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: api/transaction/transaction.proto
# plugin: python-betterproto
# This file has been @generated
from dataclasses import dataclass
from datetime import datetime
from typing import (
    TYPE_CHECKING,
    Dict,
    List,
    Optional,
)

import betterproto
import betterproto.lib.google.protobuf as betterproto_lib_google_protobuf
import grpc

from .. import (
    pfs as _pfs__,
    pps as _pps__,
)


if TYPE_CHECKING:
    import grpc


@dataclass(eq=False, repr=False)
class DeleteAllRequest(betterproto.Message):
    pass


@dataclass(eq=False, repr=False)
class TransactionRequest(betterproto.Message):
    create_repo: "_pfs__.CreateRepoRequest" = betterproto.message_field(1)
    """Exactly one of these fields should be set"""

    delete_repo: "_pfs__.DeleteRepoRequest" = betterproto.message_field(2)
    start_commit: "_pfs__.StartCommitRequest" = betterproto.message_field(3)
    finish_commit: "_pfs__.FinishCommitRequest" = betterproto.message_field(4)
    squash_commit_set: "_pfs__.SquashCommitSetRequest" = betterproto.message_field(5)
    create_branch: "_pfs__.CreateBranchRequest" = betterproto.message_field(6)
    delete_branch: "_pfs__.DeleteBranchRequest" = betterproto.message_field(7)
    update_job_state: "_pps__.UpdateJobStateRequest" = betterproto.message_field(8)
    stop_job: "_pps__.StopJobRequest" = betterproto.message_field(10)
    create_pipeline_v2: "_pps__.CreatePipelineTransaction" = betterproto.message_field(
        11
    )


@dataclass(eq=False, repr=False)
class TransactionResponse(betterproto.Message):
    commit: "_pfs__.Commit" = betterproto.message_field(1)
    """
    At most, one of these fields should be set (most responses are empty)
    """


@dataclass(eq=False, repr=False)
class Transaction(betterproto.Message):
    id: str = betterproto.string_field(1)


@dataclass(eq=False, repr=False)
class TransactionInfo(betterproto.Message):
    transaction: "Transaction" = betterproto.message_field(1)
    requests: List["TransactionRequest"] = betterproto.message_field(2)
    responses: List["TransactionResponse"] = betterproto.message_field(3)
    started: datetime = betterproto.message_field(4)
    version: int = betterproto.uint64_field(5)


@dataclass(eq=False, repr=False)
class TransactionInfos(betterproto.Message):
    transaction_info: List["TransactionInfo"] = betterproto.message_field(1)


@dataclass(eq=False, repr=False)
class BatchTransactionRequest(betterproto.Message):
    requests: List["TransactionRequest"] = betterproto.message_field(1)


@dataclass(eq=False, repr=False)
class StartTransactionRequest(betterproto.Message):
    pass


@dataclass(eq=False, repr=False)
class InspectTransactionRequest(betterproto.Message):
    transaction: "Transaction" = betterproto.message_field(1)


@dataclass(eq=False, repr=False)
class DeleteTransactionRequest(betterproto.Message):
    transaction: "Transaction" = betterproto.message_field(1)


@dataclass(eq=False, repr=False)
class ListTransactionRequest(betterproto.Message):
    pass


@dataclass(eq=False, repr=False)
class FinishTransactionRequest(betterproto.Message):
    transaction: "Transaction" = betterproto.message_field(1)


class ApiStub:

    def __init__(self, channel: "grpc.Channel"):
        self.__rpc_batch_transaction = channel.unary_unary(
            "/transaction_v2.API/BatchTransaction",
            request_serializer=BatchTransactionRequest.SerializeToString,
            response_deserializer=TransactionInfo.FromString,
        )
        self.__rpc_start_transaction = channel.unary_unary(
            "/transaction_v2.API/StartTransaction",
            request_serializer=StartTransactionRequest.SerializeToString,
            response_deserializer=Transaction.FromString,
        )
        self.__rpc_inspect_transaction = channel.unary_unary(
            "/transaction_v2.API/InspectTransaction",
            request_serializer=InspectTransactionRequest.SerializeToString,
            response_deserializer=TransactionInfo.FromString,
        )
        self.__rpc_delete_transaction = channel.unary_unary(
            "/transaction_v2.API/DeleteTransaction",
            request_serializer=DeleteTransactionRequest.SerializeToString,
            response_deserializer=betterproto_lib_google_protobuf.Empty.FromString,
        )
        self.__rpc_list_transaction = channel.unary_unary(
            "/transaction_v2.API/ListTransaction",
            request_serializer=ListTransactionRequest.SerializeToString,
            response_deserializer=TransactionInfos.FromString,
        )
        self.__rpc_finish_transaction = channel.unary_unary(
            "/transaction_v2.API/FinishTransaction",
            request_serializer=FinishTransactionRequest.SerializeToString,
            response_deserializer=TransactionInfo.FromString,
        )
        self.__rpc_delete_all = channel.unary_unary(
            "/transaction_v2.API/DeleteAll",
            request_serializer=DeleteAllRequest.SerializeToString,
            response_deserializer=betterproto_lib_google_protobuf.Empty.FromString,
        )

    def batch_transaction(
        self, *, requests: Optional[List["TransactionRequest"]] = None
    ) -> "TransactionInfo":
        requests = requests or []

        request = BatchTransactionRequest()
        if requests is not None:
            request.requests = requests

        return self.__rpc_batch_transaction(request)

    def start_transaction(self) -> "Transaction":

        request = StartTransactionRequest()

        return self.__rpc_start_transaction(request)

    def inspect_transaction(
        self, *, transaction: "Transaction" = None
    ) -> "TransactionInfo":

        request = InspectTransactionRequest()
        if transaction is not None:
            request.transaction = transaction

        return self.__rpc_inspect_transaction(request)

    def delete_transaction(
        self, *, transaction: "Transaction" = None
    ) -> "betterproto_lib_google_protobuf.Empty":

        request = DeleteTransactionRequest()
        if transaction is not None:
            request.transaction = transaction

        return self.__rpc_delete_transaction(request)

    def list_transaction(self) -> "TransactionInfos":

        request = ListTransactionRequest()

        return self.__rpc_list_transaction(request)

    def finish_transaction(
        self, *, transaction: "Transaction" = None
    ) -> "TransactionInfo":

        request = FinishTransactionRequest()
        if transaction is not None:
            request.transaction = transaction

        return self.__rpc_finish_transaction(request)

    def delete_all(self) -> "betterproto_lib_google_protobuf.Empty":

        request = DeleteAllRequest()

        return self.__rpc_delete_all(request)