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 sym_link(from_path: str, to_dir: str) → str[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]#

sym_link(from_path: str, to_dir: str) → str[source]#

upload(from_path: str, to_dir: str) → str[source]#

pydantic model ai2_kit.core.connector.SshConfig[source]#

Bases: BaseModel

Show JSON schema

{
   "$defs": {
      "SshConfig": {
         "additionalProperties": false,
         "properties": {
            "host": {
               "title": "Host",
               "type": "string"
            },
            "gateway": {
               "anyOf": [
                  {
                     "$ref": "#/$defs/SshConfig"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null
            }
         },
         "required": [
            "host"
         ],
         "title": "SshConfig",
         "type": "object"
      }
   },
   "allOf": [
      {
         "$ref": "#/$defs/SshConfig"
      }
   ]
}

Config:

extra: str = forbid

Fields:

gateway (ai2_kit.core.connector.SshConfig | None)
host (str)

field gateway: SshConfig | None = None#

field host: str [Required]#

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]#

sym_link(from_path: str, to_dir: str) → str[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