i.niyazov/frostfs-sdk-python
1
0
Fork 0
forked from TrueCloudLab/frostfs-sdk-python
Code Pull requests Activity

[#3] Added generate proto script create container method

Browse source 
Signed-off-by: Ilyas Niyazov <i.niyazov@yadro.com>
This commit is contained in:
Ilyas Niyazov 2025-03-13 14:26:01 +03:00
parent f8465e5b99
commit fba6eaaa9c
34 changed files with 547 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

37
frostfs_sdk/client/services/container.py
View file

@ -1,30 +1,41 @@
# implementation Conainer methods
from frostfs_sdk.client.models.client_environment import ClientEnvironment
from frostfs_sdk.client.services.context_accessor import ContextAccessor
from frostfs_sdk.client.parameters.container_param import ContainerCreateParam
from frostfs_sdk.client.parameters.call_context_param import CallContextParam
from frostfs_sdk.cryptography.signer import Signer
from frostfs_sdk.models.dto.container import ContainerId
import protos.models.container.service_pb2_grpc as service_pb2_grpc_container
import protos.models.container.service_pb2 as service_pb2_container
from frostfs_sdk.client.parameters.container_create import PrmContainerCreate
from frostfs_sdk.models.dto.container import ContainerId, Container
from frostfs_sdk.models.mappers.container_mapper import ContainerMapper
from frostfs_sdk.models.mappers.owner_id_mapper import OwnerIdMapper
from frostfs_sdk.models.mappers.version_mapper import VersionMapper
from frostfs_sdk.protos.models.container import service_pb2 as service_pb2_container
from frostfs_sdk.protos.models.container import service_pb2_grpc as service_pb2_grpc_container
class ContainerClient:
class ContainerClient(ContextAccessor):
def __init__(self, client_environment: ClientEnvironment):
super().__init__(client_environment)
self.container_stub = service_pb2_grpc_container.ContainerServiceStub(client_environment.channel)
self.ecdsa = client_environment.ecdsa
def create(self, prm_container_create: PrmContainerCreate) -> ContainerId:
request = self.create_put_request(prm_container_create)
def create_container(self, container_create_param: ContainerCreateParam, ctx: CallContextParam) -> ContainerId:
request = self._create_put_request(container_create_param, ctx)
response: service_pb2_container.PutResponse = self.container_stub.Put(request)
return ContainerId(value=response.body.container_id)
return ContainerId(value=response.body.container_id.value)
def create_put_request(self, prm: PrmContainerCreate):
grpc_container=ContainerMapper().to_grpc_message(prm.container)
def _create_put_request(self, param: ContainerCreateParam, ctx: CallContextParam) -> service_pb2_container.PutRequest:
"""
Creates a PUT request for creating a container.
"""
grpc_container=ContainerMapper().to_grpc_message(param.container)
if not grpc_container.owner_id:
grpc_container.owner_id = OwnerIdMapper.to_grpc_message(self.get_context.owner_id)
if not grpc_container.version:
grpc_container.version = VersionMapper.to_grpc_message(self.get_context.version)
body = service_pb2_container.PutRequest.Body(
container=grpc_container,
signature=Signer.sign_rfc6979(self.ecdsa.private_key, grpc_container)
signature=Signer.sign_message_rfc_6979(self.get_context.ecdsa, grpc_container)
)
request = service_pb2_container.PutRequest(body=body)