ai2_kit.core.connector module

ai2_kit.core.connector module#

class ai2_kit.core.connector.BaseConnector[source]#

Bases: ABC

abstract download(from_path: str, to_dir: str) → str[source]#
abstract dump_text(text: str, path: str)[source]#
abstract glob(pattern: str) → List[str][source]#
abstract run(script: str, **kwargs) → Result[source]#
abstract upload(from_path: str, to_dir: str) → str[source]#
class ai2_kit.core.connector.LocalConnector[source]#

Bases: BaseConnector

download(from_path: str, to_dir: str) → str[source]#
dump_text(text: str, path: str)[source]#
glob(pattern: str)[source]#
run(script, **kwargs)[source]#
upload(from_path: str, to_dir: str) → str[source]#
class ai2_kit.core.connector.SshConfig(*, host: str, gateway: Optional[SshConfig] = None)[source]#

Bases: BaseModel

gateway: Optional[SshConfig]#
host: str#
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'gateway': FieldInfo(annotation=Union[SshConfig, NoneType], required=False, default=None), 'host': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ai2_kit.core.connector.SshConnector(connection: Connection)[source]#

Bases: BaseConnector

download(from_path: str, to_dir: str) → str[source]#
dump_text(text: str, path: str)[source]#
classmethod from_config(config: SshConfig)[source]#
get(*args, **kwargs)[source]#
get_dir(from_dir: str, to_dir: str)[source]#
get_sftp()[source]#
glob(pattern: str)[source]#
mkdir(dir_path: str)[source]#
put(*args, **kwargs)[source]#
put_dir(from_dir: str, to_dir: str)[source]#
run(script, **kwargs) → Result[source]#
upload(from_path: str, to_dir: str) → str[source]#
ai2_kit.core.connector.get_ln_cmd(from_path: str, to_path: str)[source]#

The reason to rm -d to_path is to workaround the limit of ln. ln command cannot override existed directory, so we need to ensure to_path is not existed. Here we use -d option instead of -rf to avoid remove directory with content. The error of rm -d is suppressed as it will fail when to_path is file. -T option of ln is used to avoid some unexpected result.

ai2_kit.core.connector.safe_basename(path: str, default='')[source]#

Ensure return valid file name as basename