[#3] Added generate proto script create container method

Signed-off-by: Ilyas Niyazov <i.niyazov@yadro.com>
This commit is contained in:
Ilyas Niyazov 2025-03-25 11:12:10 +03:00
parent 19282f13cc
commit 3afa14db95
52 changed files with 1381 additions and 74 deletions

View file

@ -0,0 +1,21 @@
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
DEFAULT_GRPC_TIMEOUT = 5
class TimeUnit(Enum):
MINUTES = "MINUTES"
SECONDS = "SECONDS"
MILLISECONDS = "MILLISECONDS"
@dataclass
class CallContextParam:
timeout: int = DEFAULT_GRPC_TIMEOUT
time_unit: TimeUnit = TimeUnit.SECONDS
@classmethod
def default(cls):
return cls()

View file

@ -0,0 +1,18 @@
from dataclasses import dataclass, field
from typing import Optional, Dict
from frostfs_sdk.models.dto.container import Container
from frostfs_sdk.client.utils.session_cache import SessionToken
from frostfs_sdk.client.parameters.wait_param import WaitParam
@dataclass(frozen=True)
class ContainerCreateParam:
container: Container
wait_params: Optional[WaitParam] = None
session_token: Optional[SessionToken] = None
x_headers: Dict[str, str] = field(default_factory=dict)
def __post_init__(self):
if self.wait_params is None:
object.__setattr__(self, 'wait_params', WaitParam())

View file

@ -0,0 +1,14 @@
from dataclasses import dataclass, field
from typing import Dict, Optional
@dataclass
class CreateSessionParam:
"""
Represents parameters for creating a session.
"""
expiration: int # -1 indicates maximum expiration
x_headers: Optional[Dict[str, str]] = field(default_factory=dict)
def __init__(self, expiration: int, x_headers: Optional[Dict[str, str]] = None):
self.expiration = expiration
self.x_headers = x_headers if x_headers is not None else {}

View file

@ -0,0 +1,21 @@
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
@dataclass(frozen=True)
class WaitParam:
DEFAULT_TIMEOUT: timedelta = field(default=timedelta(seconds=120), init=False)
DEFAULT_POLL_INTERVAL: timedelta = field(default=timedelta(seconds=5), init=False)
timeout: timedelta = DEFAULT_TIMEOUT
poll_interval: timedelta = DEFAULT_POLL_INTERVAL
def __post_init__(self):
if self.timeout is None:
object.__setattr__(self, 'timeout', self.DEFAULT_TIMEOUT)
if self.poll_interval is None:
object.__setattr__(self, 'poll_interval', self.DEFAULT_POLL_INTERVAL)
def get_deadline(self) -> datetime:
return datetime.now() + self.timeout