Module pachyderm_sdk.api.transaction

Expand source code
# 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)


class ApiBase:
    def batch_transaction(
        self,
        requests: Optional[List["TransactionRequest"]],
        context: "grpc.ServicerContext",
    ) -> "TransactionInfo":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def start_transaction(self, context: "grpc.ServicerContext") -> "Transaction":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def inspect_transaction(
        self, transaction: "Transaction", context: "grpc.ServicerContext"
    ) -> "TransactionInfo":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def delete_transaction(
        self, transaction: "Transaction", context: "grpc.ServicerContext"
    ) -> "betterproto_lib_google_protobuf.Empty":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def list_transaction(self, context: "grpc.ServicerContext") -> "TransactionInfos":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def finish_transaction(
        self, transaction: "Transaction", context: "grpc.ServicerContext"
    ) -> "TransactionInfo":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def delete_all(
        self, context: "grpc.ServicerContext"
    ) -> "betterproto_lib_google_protobuf.Empty":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    __proto_path__ = "transaction_v2.API"

    @property
    def __rpc_methods__(self):
        return {
            "BatchTransaction": grpc.unary_unary_rpc_method_handler(
                self.batch_transaction,
                request_deserializer=BatchTransactionRequest.FromString,
                response_serializer=BatchTransactionRequest.SerializeToString,
            ),
            "StartTransaction": grpc.unary_unary_rpc_method_handler(
                self.start_transaction,
                request_deserializer=StartTransactionRequest.FromString,
                response_serializer=StartTransactionRequest.SerializeToString,
            ),
            "InspectTransaction": grpc.unary_unary_rpc_method_handler(
                self.inspect_transaction,
                request_deserializer=InspectTransactionRequest.FromString,
                response_serializer=InspectTransactionRequest.SerializeToString,
            ),
            "DeleteTransaction": grpc.unary_unary_rpc_method_handler(
                self.delete_transaction,
                request_deserializer=DeleteTransactionRequest.FromString,
                response_serializer=DeleteTransactionRequest.SerializeToString,
            ),
            "ListTransaction": grpc.unary_unary_rpc_method_handler(
                self.list_transaction,
                request_deserializer=ListTransactionRequest.FromString,
                response_serializer=ListTransactionRequest.SerializeToString,
            ),
            "FinishTransaction": grpc.unary_unary_rpc_method_handler(
                self.finish_transaction,
                request_deserializer=FinishTransactionRequest.FromString,
                response_serializer=FinishTransactionRequest.SerializeToString,
            ),
            "DeleteAll": grpc.unary_unary_rpc_method_handler(
                self.delete_all,
                request_deserializer=DeleteAllRequest.FromString,
                response_serializer=DeleteAllRequest.SerializeToString,
            ),
        }

Sub-modules

pachyderm_sdk.api.transaction.extension

Handwritten classes/methods that augment the existing Transaction API.

Classes

class DeleteAllRequest

DeleteAllRequest()

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

Ancestors

  • betterproto.Message
  • abc.ABC
class TransactionRequest (create_repo: _pfs__.CreateRepoRequest = None, delete_repo: _pfs__.DeleteRepoRequest = None, start_commit: _pfs__.StartCommitRequest = None, finish_commit: _pfs__.FinishCommitRequest = None, squash_commit_set: _pfs__.SquashCommitSetRequest = None, create_branch: _pfs__.CreateBranchRequest = None, delete_branch: _pfs__.DeleteBranchRequest = None, update_job_state: _pps__.UpdateJobStateRequest = None, stop_job: _pps__.StopJobRequest = None, create_pipeline_v2: _pps__.CreatePipelineTransaction = None)

TransactionRequest(create_repo: '_pfs__.CreateRepoRequest' = None, delete_repo: '_pfs__.DeleteRepoRequest' = None, start_commit: '_pfs__.StartCommitRequest' = None, finish_commit: '_pfs__.FinishCommitRequest' = None, squash_commit_set: '_pfs__.SquashCommitSetRequest' = None, create_branch: '_pfs__.CreateBranchRequest' = None, delete_branch: '_pfs__.DeleteBranchRequest' = None, update_job_state: '_pps__.UpdateJobStateRequest' = None, stop_job: '_pps__.StopJobRequest' = None, create_pipeline_v2: '_pps__.CreatePipelineTransaction' = None)

Expand source code
@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
    )

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var create_repoCreateRepoRequest

Exactly one of these fields should be set

var delete_repoDeleteRepoRequest
var start_commitStartCommitRequest
var finish_commitFinishCommitRequest
var squash_commit_setSquashCommitSetRequest
var create_branchCreateBranchRequest
var delete_branchDeleteBranchRequest
var update_job_stateUpdateJobStateRequest
var stop_jobStopJobRequest
var create_pipeline_v2CreatePipelineTransaction
class TransactionResponse (commit: _pfs__.Commit = None)

TransactionResponse(commit: '_pfs__.Commit' = None)

Expand source code
@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)
    """

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var commitCommit

At most, one of these fields should be set (most responses are empty)

class Transaction (id: str = None)

Transaction(id: str = None)

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

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var id : str
class TransactionInfo (transaction: Transaction = None, requests: List[ForwardRef('TransactionRequest')] = None, responses: List[ForwardRef('TransactionResponse')] = None, started: datetime.datetime = None, version: int = None)

TransactionInfo(transaction: 'Transaction' = None, requests: List[ForwardRef('TransactionRequest')] = None, responses: List[ForwardRef('TransactionResponse')] = None, started: datetime.datetime = None, version: int = None)

Expand source code
@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)

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var transactionTransaction
var requests : List[TransactionRequest]
var responses : List[TransactionResponse]
var started : datetime.datetime
var version : int
class TransactionInfos (transaction_info: List[ForwardRef('TransactionInfo')] = None)

TransactionInfos(transaction_info: List[ForwardRef('TransactionInfo')] = None)

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

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var transaction_info : List[TransactionInfo]
class BatchTransactionRequest (requests: List[ForwardRef('TransactionRequest')] = None)

BatchTransactionRequest(requests: List[ForwardRef('TransactionRequest')] = None)

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

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var requests : List[TransactionRequest]
class StartTransactionRequest

StartTransactionRequest()

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

Ancestors

  • betterproto.Message
  • abc.ABC
class InspectTransactionRequest (transaction: Transaction = None)

InspectTransactionRequest(transaction: 'Transaction' = None)

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

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var transactionTransaction
class DeleteTransactionRequest (transaction: Transaction = None)

DeleteTransactionRequest(transaction: 'Transaction' = None)

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

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var transactionTransaction
class ListTransactionRequest

ListTransactionRequest()

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

Ancestors

  • betterproto.Message
  • abc.ABC
class FinishTransactionRequest (transaction: Transaction = None)

FinishTransactionRequest(transaction: 'Transaction' = None)

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

Ancestors

  • betterproto.Message
  • abc.ABC

Class variables

var transactionTransaction
class ApiStub (channel: grpc.Channel)
Expand source code
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)

Subclasses

Methods

def batch_transaction(self, *, requests: Optional[List[ForwardRef('TransactionRequest')]] = None) ‑> TransactionInfo
Expand source code
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
Expand source code
def start_transaction(self) -> "Transaction":
    request = StartTransactionRequest()

    return self.__rpc_start_transaction(request)
def inspect_transaction(self, *, transaction: Transaction = None) ‑> TransactionInfo
Expand source code
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
Expand source code
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
Expand source code
def list_transaction(self) -> "TransactionInfos":
    request = ListTransactionRequest()

    return self.__rpc_list_transaction(request)
def finish_transaction(self, *, transaction: Transaction = None) ‑> TransactionInfo
Expand source code
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
Expand source code
def delete_all(self) -> "betterproto_lib_google_protobuf.Empty":
    request = DeleteAllRequest()

    return self.__rpc_delete_all(request)
class ApiBase
Expand source code
class ApiBase:
    def batch_transaction(
        self,
        requests: Optional[List["TransactionRequest"]],
        context: "grpc.ServicerContext",
    ) -> "TransactionInfo":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def start_transaction(self, context: "grpc.ServicerContext") -> "Transaction":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def inspect_transaction(
        self, transaction: "Transaction", context: "grpc.ServicerContext"
    ) -> "TransactionInfo":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def delete_transaction(
        self, transaction: "Transaction", context: "grpc.ServicerContext"
    ) -> "betterproto_lib_google_protobuf.Empty":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def list_transaction(self, context: "grpc.ServicerContext") -> "TransactionInfos":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def finish_transaction(
        self, transaction: "Transaction", context: "grpc.ServicerContext"
    ) -> "TransactionInfo":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def delete_all(
        self, context: "grpc.ServicerContext"
    ) -> "betterproto_lib_google_protobuf.Empty":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    __proto_path__ = "transaction_v2.API"

    @property
    def __rpc_methods__(self):
        return {
            "BatchTransaction": grpc.unary_unary_rpc_method_handler(
                self.batch_transaction,
                request_deserializer=BatchTransactionRequest.FromString,
                response_serializer=BatchTransactionRequest.SerializeToString,
            ),
            "StartTransaction": grpc.unary_unary_rpc_method_handler(
                self.start_transaction,
                request_deserializer=StartTransactionRequest.FromString,
                response_serializer=StartTransactionRequest.SerializeToString,
            ),
            "InspectTransaction": grpc.unary_unary_rpc_method_handler(
                self.inspect_transaction,
                request_deserializer=InspectTransactionRequest.FromString,
                response_serializer=InspectTransactionRequest.SerializeToString,
            ),
            "DeleteTransaction": grpc.unary_unary_rpc_method_handler(
                self.delete_transaction,
                request_deserializer=DeleteTransactionRequest.FromString,
                response_serializer=DeleteTransactionRequest.SerializeToString,
            ),
            "ListTransaction": grpc.unary_unary_rpc_method_handler(
                self.list_transaction,
                request_deserializer=ListTransactionRequest.FromString,
                response_serializer=ListTransactionRequest.SerializeToString,
            ),
            "FinishTransaction": grpc.unary_unary_rpc_method_handler(
                self.finish_transaction,
                request_deserializer=FinishTransactionRequest.FromString,
                response_serializer=FinishTransactionRequest.SerializeToString,
            ),
            "DeleteAll": grpc.unary_unary_rpc_method_handler(
                self.delete_all,
                request_deserializer=DeleteAllRequest.FromString,
                response_serializer=DeleteAllRequest.SerializeToString,
            ),
        }

Methods

def batch_transaction(self, requests: Optional[List[ForwardRef('TransactionRequest')]], context: grpc.ServicerContext) ‑> TransactionInfo
Expand source code
def batch_transaction(
    self,
    requests: Optional[List["TransactionRequest"]],
    context: "grpc.ServicerContext",
) -> "TransactionInfo":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")
def start_transaction(self, context: grpc.ServicerContext) ‑> Transaction
Expand source code
def start_transaction(self, context: "grpc.ServicerContext") -> "Transaction":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")
def inspect_transaction(self, transaction: Transaction, context: grpc.ServicerContext) ‑> TransactionInfo
Expand source code
def inspect_transaction(
    self, transaction: "Transaction", context: "grpc.ServicerContext"
) -> "TransactionInfo":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")
def delete_transaction(self, transaction: Transaction, context: grpc.ServicerContext) ‑> betterproto.lib.google.protobuf.Empty
Expand source code
def delete_transaction(
    self, transaction: "Transaction", context: "grpc.ServicerContext"
) -> "betterproto_lib_google_protobuf.Empty":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")
def list_transaction(self, context: grpc.ServicerContext) ‑> TransactionInfos
Expand source code
def list_transaction(self, context: "grpc.ServicerContext") -> "TransactionInfos":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")
def finish_transaction(self, transaction: Transaction, context: grpc.ServicerContext) ‑> TransactionInfo
Expand source code
def finish_transaction(
    self, transaction: "Transaction", context: "grpc.ServicerContext"
) -> "TransactionInfo":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")
def delete_all(self, context: grpc.ServicerContext) ‑> betterproto.lib.google.protobuf.Empty
Expand source code
def delete_all(
    self, context: "grpc.ServicerContext"
) -> "betterproto_lib_google_protobuf.Empty":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")