cmdbox.app package

Subpackages

Submodules

cmdbox.app.app module

class cmdbox.app.app.CmdBoxApp(appcls=None, ver=<module 'cmdbox.version' from 'C:\\Users\\hama\\OneDrive\\デスクトップ\\devenv\\cmdbox\\cmdbox\\version.py'>, cli_features_packages: ~typing.List[str] = None, cli_features_prefix: ~typing.List[str] = None)[source]

Bases: object

static getInstance(appcls=None, ver=<module 'cmdbox.version' from 'C:\\Users\\hama\\OneDrive\\デスクトップ\\devenv\\cmdbox\\cmdbox\\version.py'>, cli_features_packages: ~typing.List[str] = None, cli_features_prefix: ~typing.List[str] = None)[source]
main(args_list: list = None, file_dict: dict = {}, webcall: bool = False)[source]

コマンドライン引数を処理し、サーバーまたはクライアントを起動し、コマンドを実行する。

cmdbox.app.app.main(args_list: list = None)[source]

cmdbox.app.client module

class cmdbox.app.client.Client(logger: Logger, redis_host: str = 'localhost', redis_port: int = 6379, redis_password: str = None, svname: str = 'server')[source]

Bases: object

file_copy(from_path: str, to_path: str, orverwrite: bool = False, scope: str = 'client', client_data: Path = None, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上のファイルをコピーする

Parameters:
  • from_path (Path) -- コピー元のファイルパス

  • to_path (Path) -- コピー先のファイルパス

  • orverwrite (bool, optional) -- 上書きするかどうか. Defaults to False.

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

file_download(svpath: str, download_file: Path, scope: str = 'client', client_data: Path = None, rpath: str = '', img_thumbnail: float = 0.0, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上のファイルをダウンロードする

Parameters:
  • svpath (Path) -- サーバー上のファイルパス

  • download_file (Path) -- ローカルのファイルパス

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • rpath (str, optional) -- リクエストパス. Defaults to "".

  • img_thumbnail (float, optional) -- サムネイル画像のサイズ. Defaults to 0.0.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

ダウンロードファイルの内容

Return type:

bytes

file_list(svpath: str, recursive: bool, scope: str = 'client', client_data: Path = None, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上のファイルリストを取得する

Parameters:
  • svpath (Path) -- サーバー上のファイルパス

  • recursive (bool) -- 再帰的に取得するかどうか

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

file_mkdir(svpath: str, scope: str = 'client', client_data: Path = None, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上にディレクトリを作成する

Parameters:
  • svpath (Path) -- サーバー上のディレクトリパス

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

file_move(from_path: str, to_path: str, scope: str = 'client', client_data: Path = None, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上のファイルを移動する

Parameters:
  • from_path (Path) -- 移動元のファイルパス

  • to_path (Path) -- 移動先のファイルパス

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

file_remove(svpath: str, scope: str = 'client', client_data: Path = None, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上のファイルを削除する

Parameters:
  • svpath (Path) -- サーバー上のファイルパス

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

file_rmdir(svpath: str, scope: str = 'client', client_data: Path = None, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上のディレクトリを削除する

Parameters:
  • svpath (Path) -- サーバー上のディレクトリパス

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

file_upload(svpath: str, upload_file: Path, scope: str = 'client', client_data: Path = None, mkdir: bool = False, orverwrite: bool = False, retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバー上にファイルをアップロードする

Parameters:
  • svpath (Path) -- サーバー上のファイルパス

  • upload_file (Path) -- ローカルのファイルパス

  • scope (str, optional) -- 参照先のスコープ. Defaults to "client".

  • mkdir (bool, optional) -- ディレクトリを作成するかどうか. Defaults to False.

  • orverwrite (bool, optional) -- 上書きするかどうか. Defaults to False.

  • client_data (Path, optional) -- ローカルを参照させる場合のデータフォルダ. Defaults to None.

  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

server_info(retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

サーバーの情報を取得する

Parameters:
  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

stop_server(retry_count: int = 3, retry_interval: int = 5, timeout: int = 60)[source]

Redisサーバーを停止する

Parameters:
  • retry_count (int, optional) -- リトライ回数. Defaults to 3.

  • retry_interval (int, optional) -- リトライ間隔. Defaults to 5.

  • timeout (int, optional) -- タイムアウト時間. Defaults to 60.

Returns:

Redisサーバーからの応答

Return type:

dict

cmdbox.app.common module

cmdbox.app.common.check_fname(fname: str) bool[source]

ファイル名に使えない文字が含まれているかどうかをチェックします。

Parameters:

fname (str) -- ファイル名

Returns:

Trueの場合は使えない文字が含まれている

Return type:

bool

cmdbox.app.common.chopdq(target: str)[source]

" "で囲まれた文字列を取り除きます。 targetにNoneが指定された場合はNoneを返します。

Parameters:

target (str) -- 対象文字列

Returns:

取り除かれた文字列

Return type:

str

cmdbox.app.common.cmd(cmd: str, logger: Logger, slise: int = 100, newenv: Dict = None) Tuple[int, str, str][source]

コマンドを実行します。

Parameters:
  • cmd (str) -- 実行するコマンド

  • logger (logging.Logger) -- ロガー

  • slise (int, optional) -- 出力文字列の最大長. Defaults to 100

  • newenv (dict) -- 上書きしたい環境変数

Returns:

コマンドの戻り値と出力とコマンド

Return type:

Tuple[int, str, str]

cmdbox.app.common.copy_sample(data: ~pathlib.Path, ver=<module 'cmdbox.version' from 'C:\\Users\\hama\\OneDrive\\デスクトップ\\devenv\\cmdbox\\cmdbox\\version.py'>)[source]

サンプルデータをコピーします。

Parameters:
  • data (Path) -- データディレクトリ

  • ver (version, optional) -- バージョン. Defaults to version

cmdbox.app.common.create_console(stderr: bool = False, file=None) Console[source]

コンソールを生成します。

Parameters:
  • stderr (bool, optional) -- 標準エラー出力を使用するかどうか. Defaults to False.

  • file (file, optional) -- 出力先のファイル. Defaults to None.

Returns:

コンソール

Return type:

Console

cmdbox.app.common.create_log_handler(stderr: bool = False, fmt: str = '[%(asctime)s] %(levelname)s - %(message)s', datefmt: str = '%Y-%m-%d %H:%M:%S') Handler[source]

ログハンドラを生成します。

Parameters:
  • stderr (bool, optional) -- 標準エラー出力を使用するかどうか. Defaults to False.

  • fmt (str, optional) -- ログフォーマット. Defaults to '[%(asctime)s] %(levelname)s - %(message)s'.

  • datefmt (str, optional) -- 日時フォーマット. Defaults to '%Y-%m-%d %H:%M:%S'.

Returns:

ログハンドラ

Return type:

logging.Handler

cmdbox.app.common.decrypt(enc_message: str, password: str) str[source]

メッセージを復号化します。

Parameters:
  • enc_message (str) -- 暗号化されたメッセージ

  • password (str) -- パスワード

Returns:

復号化されたメッセージ。失敗した場合はNone

Return type:

str

cmdbox.app.common.default_json_enc(o) Any[source]
cmdbox.app.common.default_logger(debug: bool = False, ver=<module 'cmdbox.version' from 'C:\\Users\\hama\\OneDrive\\デスクトップ\\devenv\\cmdbox\\cmdbox\\version.py'>, webcall: bool = False) Logger[source]

デフォルトのロガーを生成します。

Parameters:
  • debug (bool, optional) -- デバッグモード. Defaults to False.

  • ver (version, optional) -- バージョン. Defaults to version.

  • webcall (bool, optional) -- WebAPIからの呼出しの場合はTrue. setHandlerを削除します。. Defaults to False.

Returns:

ロガー

Return type:

logging.Logger

cmdbox.app.common.download_file(url: str, save_path: Path) Path[source]

ファイルをダウンロードします。

Parameters:
  • url (str) -- ダウンロードするファイルのURL

  • save_path (Path) -- 保存先のファイルパス

Returns:

保存したファイルのパス

Return type:

Path

cmdbox.app.common.encrypt(message: str, password: str) str[source]

メッセージを暗号化します。

Parameters:
  • message (str) -- メッセージ

  • password (str) -- パスワード

Returns:

暗号化されたメッセージ

Return type:

str

cmdbox.app.common.exec_sync(func, *args, **kwargs) Any[source]

" 指定された関数が非同期関数であっても同期的に実行します。

Parameters:
  • func (function) -- 関数

  • args (Any) -- 引数

  • kwargs (Any) -- キーワード引数

Returns:

関数の戻り値

Return type:

Any

cmdbox.app.common.get_tzoffset_str() str[source]

タイムゾーンのオフセットを取得します

Returns:

タイムゾーンのオフセット

Return type:

str

cmdbox.app.common.getopt(opt: dict, key: str, preval=None, defval=None, withset=False) any[source]

コマンドラインオプションから指定されたキーの値を取得します。

Parameters:
  • opt (dict) -- 読み込んだコマンドラインオプション

  • key (str) -- キー

  • preval (Any, optional) -- 引数で指定されたコマンドラインオプション. Defaults to None.

  • defval (Any, optional) -- デフォルト値. Defaults to None.

  • withset (bool, optional) -- optに引数のコマンドラインオプションを設定するかどうか. Defaults to False.

Returns:

取得した値

Return type:

Any

cmdbox.app.common.hash_password(password: str, hash: str) str[source]

パスワードをハッシュ化します。

Parameters:
  • password (str) -- パスワード

  • hash (str) -- ハッシュアルゴリズム

Returns:

ハッシュ化されたパスワード

Return type:

str

cmdbox.app.common.is_event_loop_running() bool[source]

イベントループが実行中かどうかを取得します。

Returns:

イベントループが実行中の場合はTrue

Return type:

bool

cmdbox.app.common.load_config(mode: str, debug: bool = False, data=WindowsPath('C:/Users/hama'), webcall: bool = False, ver=<module 'cmdbox.version' from 'C:\\Users\\hama\\OneDrive\\デスクトップ\\devenv\\cmdbox\\cmdbox\\version.py'>) Tuple[Logger, dict][source]

指定されたモードのロガーと設定を読み込みます。

Parameters:
  • mode (str) -- モード名

  • debug (bool, optional) -- デバッグモード. Defaults to False

  • data (Path, optional) -- データディレクトリ. Defaults to HOME_DIR.

  • webcall (bool, optional) -- WebAPIからの呼出しの場合はTrue. setHandlerを削除します。. Defaults to False.

  • appid (str, optional) -- アプリケーションID. Defaults to version.__appid__.

Returns:

ロガー config (dict): 設定

Return type:

logger (logging.Logger)

cmdbox.app.common.load_yml(yml_path: Path) dict[source]

YAMLファイルを読み込みます。

Parameters:

yml_path (Path) -- YAMLファイルのパス

Returns:

読み込んだYAMLファイルの内容

Return type:

dict

cmdbox.app.common.loadopt(opt_path: str, webmode: bool = False) dict[source]

JSON形式のファイルからコマンドラインオプションを読み込みます。

Parameters:
  • opt_path (str) -- 読み込むファイルパス

  • webmode (bool, optional) -- Webモードからの呼び出し. Defaults to False

Returns:

読み込んだコマンドラインオプション

Return type:

dict

cmdbox.app.common.loaduser(user_path: str) dict[source]

JSON形式のユーザーファイルを読み込みます。

Parameters:

user_path (str) -- ユーザーファイルのパス

Returns:

読み込んだユーザーファイル

Return type:

dict

cmdbox.app.common.mkdirs(dir_path: Path)[source]

ディレクトリを中間パスも含めて作成します。

Parameters:

dir_path (Path) -- 作成するディレクトリのパス

Returns:

作成したディレクトリのパス

Return type:

Path

cmdbox.app.common.mklogdir(data: Path) Path[source]

ログディレクトリを作成します。

Parameters:

logdir (Path, optional) -- ログディレクトリのパス. Defaults to Path.cwd()/'log'.

Returns:

作成したログディレ作成したログディレクトリのパス

cmdbox.app.common.print_format(data: dict, format: bool, tm: float, output_json: str = None, output_json_append: bool = False, stdout: bool = True, tablefmt: str = 'github', pf: List[Dict[str, float]] = [])[source]

データを指定されたフォーマットで出力します。

Parameters:
  • data (dict) -- 出力するデータ

  • format (bool) -- フォーマットするかどうか

  • tm (float) -- 処理時間

  • output_json (str, optional) -- JSON形式で出力するファイルパス. Defaults to None.

  • output_json_append (bool, optional) -- JSON形式で出力するファイルパス. Defaults to False.

  • stdout (bool, optional) -- 標準出力に出力するかどうか. Defaults to True.

  • tablefmt (str, optional) -- テーブルのフォーマット. Defaults to 'github'.

  • pf (List[Dict[str, float]], optional) -- パフォーマンス情報. Defaults to [].

Returns:

生成された文字列

Return type:

str

cmdbox.app.common.random_string(size: int = 16, chars='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789') str[source]

ランダムな文字列を生成します。

Parameters:
  • size (int, optional) -- 文字列の長さ. Defaults to 16.

  • chars ([type], optional) -- 使用する文字. Defaults to string.ascii_uppercase + string.digits.

Returns:

生成された文字列

Return type:

str

cmdbox.app.common.reset_logger(name: str, stderr: bool = False, fmt: str = '[%(asctime)s] %(levelname)s - %(message)s', datefmt: str = '%Y-%m-%d %H:%M:%S') None[source]

指定されたロガーのハンドラをクリアし、新しいハンドラを追加します。 :param name: ロガーの名前 :type name: str :param stderr: 標準エラー出力を使用するかどうか. Defaults to False. :type stderr: bool, optional :param fmt: ログフォーマット. Defaults to '[%(asctime)s] %(levelname)s - %(message)s'. :type fmt: str, optional :param datefmt: 日時フォーマット. Defaults to '%Y-%m-%d %H:%M:%S'. :type datefmt: str, optional

cmdbox.app.common.rmdirs(dir_path: Path, ignore_errors: bool = True)[source]

ディレクトリをサブディレクトリ含めて削除します。

Parameters:

dir_path (Path) -- 削除するディレクトリのパス

cmdbox.app.common.safe_fname(fname: str) str[source]

ファイル名に使えない文字を置換します。

Parameters:

fname (str) -- ファイル名

Returns:

置換後のファイル名

Return type:

str

cmdbox.app.common.save_yml(yml_path: Path, data: dict) None[source]

YAMLファイルに書き込みます。

Parameters:
  • yml_path (Path) -- YAMLファイルのパス

  • data (dict) -- 書き込むデータ

cmdbox.app.common.saveopt(opt: dict, opt_path: Path, webmode: bool = False) None[source]

コマンドラインオプションをJSON形式でファイルに保存します。

Parameters:
  • opt (dict) -- コマンドラインオプション

  • opt_path (Path) -- 保存先のファイルパス

  • webmode (bool, optional) -- Webモードからの呼び出し. Defaults to False.

cmdbox.app.common.saveuser(user_data: dict, user_path: Path) None[source]

ユーザーファイルを保存します。

Parameters:
  • user_data (dict) -- ユーザーファイル

  • user_path (Path) -- 保存先のファイルパス

cmdbox.app.common.set_debug(logger: Logger, debug: bool = False) None[source]

ロガーのデバッグモードを設定します。

Parameters:
  • logger (logging.Logger) -- ロガー

  • debug (bool, optional) -- デバッグモード. Defaults to False.

cmdbox.app.common.to_str(o, slise=-1)[source]

cmdbox.app.edge module

class cmdbox.app.edge.Edge(logger: Logger, data: str, appcls=None, ver=None)[source]

Bases: object

configure(edge_mode: str, edge_cmd: str, args: Namespace, tm: float, pf: List[Dict[str, float]] = []) Dict[str, str][source]

端末モードの設定を行います

Parameters:
  • edge_mode (str) -- edgeモード

  • edge_cmd (str) -- edgeコマンド

Returns:

メッセージ

Return type:

Dict[str, str]

exec_pipe(opt: Dict[str, str]) Dict[str, str][source]

パイプを実行します

Parameters:

opt (Dict[str, str]) -- パイプオプション

Returns:

メッセージ

Return type:

Dict[str, str]

load_user_info() Tuple[int, Dict[str, Any]][source]
signin(auth_type: str, user: str, password: str, apikey: str, oauth2: str, oauth2_port: int, oauth2_tenant_id: str, oauth2_client_id: str, oauth2_client_secret: str, oauth2_timeout: int, saml: str, saml_port: int, saml_tenant_id: str, saml_timeout: int) Tuple[int, Dict[str, Any]][source]

サインインを行います

Parameters:
  • auth_type (str) -- 認証タイプ

  • user (str) -- ユーザー名

  • password (str) -- パスワード

  • apikey (str) -- APIキー

  • oauth2 (str) -- OAuth2

  • oauth2_port (int) -- OAuth2ポート

  • oauth2_tenant_id (str) -- OAuth2テナントID

  • oauth2_client_id (str) -- OAuth2クライアントID

  • oauth2_client_secret (str) -- OAuth2クライアントシークレット

  • oauth2_timeout (int) -- OAuth2タイムアウト

  • saml (str) -- SAML

  • saml_port (int) -- SAMLポート

  • saml_tenant_id (str) -- SAMLテナントID

  • saml_timeout (int) -- SAMLタイムアウト

Returns:

終了コード, メッセージ

Return type:

Tuple[int, Dict[str, Any]]

site_request(func, path: str, headers: Dict[str, str] = None, data: Any = None, allow_redirects: bool = False, ok_status: List[int] = [200]) Tuple[int, Any, Dict[str, str]][source]
start(resignin: bool = False) Dict[str, str][source]

Edgeを起動します

Parameters:

resignin (bool) -- サインインを再実行する

Returns:

メッセージ

Return type:

Dict[str, str]

start_tray() Dict[str, str][source]
stop_jobs(no_notify: bool) None[source]

cmdbox.app.feature module

class cmdbox.app.feature.Feature(appcls, ver)[source]

Bases: object

DEFAULT_CAPTURE_MAXSIZE: int = 10485760
RESP_ERROR: int = 2
RESP_SCCESS: int = 0
RESP_WARN: int = 1
USE_REDIS_FALSE: int = -1
USE_REDIS_MEIGHT: int = 0
USE_REDIS_TRUE: int = 1
apprun(logger: Logger, args: Namespace, tm: float, pf: List[Dict[str, float]]) Tuple[int, Dict[str, Any], Any][source]

この機能の実行を行います

Parameters:
  • logger (logging.Logger) -- ロガー

  • args (argparse.Namespace) -- 引数

  • tm (float) -- 実行開始時間

  • pf (List[Dict[str, float]]) -- 呼出元のパフォーマンス情報

Returns:

終了コード, 結果, オブジェクト

Return type:

Tuple[int, Dict[str, Any], Any]

default_host: str = 'localhost'
default_pass: str = 'password'
default_port: int = 6379
default_svname: str = 'server'
edgerun(opt: Dict[str, Any], tool: Tool, logger: Logger, timeout: int, prevres: Any = None)[source]

この機能のエッジ側の実行を行います

Parameters:
  • opt (Dict[str, Any]) -- オプション

  • tool (edge_tool.Tool) -- 通知関数などedge側のUI操作を行うためのクラス

  • logger (logging.Logger) -- ロガー

  • timeout (int) -- タイムアウト時間

  • prevres (Any) -- 前コマンドの結果。pipeline実行の実行結果を参照する時に使用します。

Yields:

Tuple[int, Dict[str, Any]] -- 終了コード, 結果

get_cmd() str[source]

この機能のコマンドを返します

Returns:

コマンド

Return type:

str

get_mode() str | List[str][source]

この機能のモードを返します

Returns:

モード

Return type:

Union[str, List[str]]

get_option() Dict[str, Any][source]

この機能のオプションを返します

Returns:

オプション

Return type:

Dict[str, Any]

get_svcmd()[source]

この機能のサーバー側のコマンドを返します

Returns:

サーバー側のコマンド

Return type:

str

is_cluster_redirect()[source]

クラスター宛のメッセージの場合、メッセージを転送するかどうかを返します

Returns:

メッセージを転送する場合はTrue

Return type:

bool

svrun(data_dir: Path, logger: Logger, redis_cli: RedisClient, msg: List[str], sessions: Dict[str, Dict[str, Any]]) int[source]

この機能のサーバー側の実行を行います

Parameters:
  • data_dir (Path) -- データディレクトリ

  • logger (logging.Logger) -- ロガー

  • redis_cli (redis_client.RedisClient) -- Redisクライアント

  • msg (List[str]) -- 受信メッセージ

  • sessions (Dict[str, Dict[str, Any]]) -- セッション情報

Returns:

終了コード

Return type:

int

class cmdbox.app.feature.OneshotEdgeFeature(appcls, ver)[source]

Bases: Feature

一度だけ実行するエッジ機能の基底クラス

edgerun(opt: Dict[str, Any], tool: Tool, logger: Logger, timeout: int, prevres: Any = None)[source]

この機能のエッジ側の実行を行います

Parameters:
  • opt (Dict[str, Any]) -- オプション

  • tool (edge_tool.Tool) -- 通知関数などedge側のUI操作を行うためのクラス

  • logger (logging.Logger) -- ロガー

  • timeout (int) -- タイムアウト時間

  • prevres (Any) -- 前コマンドの結果。pipeline実行の実行結果を参照する時に使用します。

Yields:

Tuple[int, Dict[str, Any]] -- 終了コード, 結果

class cmdbox.app.feature.OneshotNotifyEdgeFeature(appcls, ver)[source]

Bases: OneshotEdgeFeature

実行結果の通知を行うエッジ機能の基底クラス

edgerun(opt: Dict[str, Any], tool: Tool, logger: Logger, timeout: int, prevres: Any = None)[source]

この機能のエッジ側の実行を行います

Parameters:
  • opt (Dict[str, Any]) -- オプション

  • tool (edge_tool.Tool) -- 通知関数などedge側のUI操作を行うためのクラス

  • logger (logging.Logger) -- ロガー

  • timeout (int) -- タイムアウト時間

  • prevres (Any) -- 前コマンドの結果。pipeline実行の実行結果を参照する時に使用します。

Yields:

Tuple[int, Dict[str, Any]] -- 終了コード, 結果

class cmdbox.app.feature.OneshotResultEdgeFeature(appcls, ver)[source]

Bases: ResultEdgeFeature

一度だけ実行結果をWebブラウザで表示するエッジ機能の基底クラス

edgerun(opt: Dict[str, Any], tool: Tool, logger: Logger, timeout: int, prevres: Any = None)[source]

この機能のエッジ側の実行を行います

Parameters:
  • opt (Dict[str, Any]) -- オプション

  • tool (edge_tool.Tool) -- 通知関数などedge側のUI操作を行うためのクラス

  • logger (logging.Logger) -- ロガー

  • timeout (int) -- タイムアウト時間

  • prevres (Any) -- 前コマンドの結果。pipeline実行の実行結果を参照する時に使用します。

Yields:

Tuple[int, Dict[str, Any]] -- 終了コード, 結果

class cmdbox.app.feature.ResultEdgeFeature(appcls, ver)[source]

Bases: Feature

実行結果をWebブラウザで表示するエッジ機能の基底クラス

edgerun(opt: Dict[str, Any], tool: Tool, logger: Logger, timeout: int, prevres: Any = None)[source]

この機能のエッジ側の実行を行います

Parameters:
  • opt (Dict[str, Any]) -- オプション

  • tool (edge_tool.Tool) -- 通知関数などedge側のUI操作を行うためのクラス

  • logger (logging.Logger) -- ロガー

  • timeout (int) -- タイムアウト時間

  • prevres (Any) -- 前コマンドの結果。pipeline実行の実行結果を参照する時に使用します。

Yields:

Tuple[int, Dict[str, Any]] -- 終了コード, 結果

class cmdbox.app.feature.UnsupportEdgeFeature(appcls, ver)[source]

Bases: Feature

サポートされていないエッジ機能の基底クラス

edgerun(opt: Dict[str, Any], tool: Tool, logger: Logger, timeout: int, prevres: Any = None)[source]

この機能のエッジ側の実行を行います

Parameters:
  • opt (Dict[str, Any]) -- オプション

  • tool (edge_tool.Tool) -- 通知関数などedge側のUI操作を行うためのクラス

  • logger (logging.Logger) -- ロガー

  • timeout (int) -- タイムアウト時間

  • prevres (Any) -- 前コマンドの結果。pipeline実行の実行結果を参照する時に使用します。

Yields:

Tuple[int, Dict[str, Any]] -- 終了コード, 結果

class cmdbox.app.feature.WebFeature(appcls=None, ver=<module 'cmdbox.version' from 'C:\\Users\\hama\\OneDrive\\デスクトップ\\devenv\\cmdbox\\cmdbox\\version.py'>)[source]

Bases: object

DEFAULT_401_MESSAGE: str = 'Unauthorized operation. Please sign in again as an authorized user.'
DEFAULT_CAPTURE_MAXSIZE: int = 10485760
USE_REDIS_FALSE: int = -1
USE_REDIS_MEIGHT: int = 0
USE_REDIS_TRUE: int = 1
aboutmenu(web: Web) Dict[str, Any][source]

Aboutメニューの情報を返します

Parameters:

web (Web) -- Webオブジェクト

Returns:

Aboutメニュー情報

Return type:

Dict[str, Any]

Notes

以下は返されるJSONのサンプル:

{
    'filer': {
        'html': 'Filer',
        'href': 'filer',
        'target': '_blank',
        'css_class': 'dropdown-item',
        'onclick': 'alert("filer")'
    }
}
filemenu(web: Web) Dict[str, Any][source]

ファイルメニューの情報を返します

Parameters:

web (Web) -- Webオブジェクト

Returns:

fileメニュー情報

Return type:

Dict[str, Any]

Notes

以下は返されるJSONのサンプル:

{
    'filer': {
        'html': 'Filer',
        'href': 'filer',
        'target': '_blank',
        'css_class': 'dropdown-item'
        'onclick': 'alert("filer")'
    }
}
route(web: Web, app: FastAPI) None[source]

webモードのルーティングを設定します

Parameters:
  • web (Web) -- Webオブジェクト

  • app (FastAPI) -- FastAPIオブジェクト

toolmenu(web: Web) Dict[str, Any][source]

ツールメニューの情報を返します

Parameters:

web (Web) -- Webオブジェクト

Returns:

ツールメニュー情報

Return type:

Dict[str, Any]

Notes

以下は返されるJSONのサンプル:

{
    'filer': {
        'html': 'Filer',
        'href': 'filer',
        'target': '_blank',
        'css_class': 'dropdown-item',
        'onclick': 'alert("filer")'
    }
}
viewmenu(web: Web) Dict[str, Any][source]

Viewメニューの情報を返します

Parameters:

web (Web) -- Webオブジェクト

Returns:

Viewメニュー情報

Return type:

Dict[str, Any]

Notes

以下は返されるJSONのサンプル:

{
    'filer': {
        'html': 'Filer',
        'href': 'filer',
        'target': '_blank',
        'css_class': 'dropdown-item',
        'onclick': 'alert("filer")'
    }
}

cmdbox.app.filer module

class cmdbox.app.filer.Filer(data_dir: Path, logger: Logger)[source]

Bases: object

RESP_ERROR: int = 2
RESP_SCCESS: int = 0
RESP_WARN: int = 1
file_copy(from_path: str, to_path: str, orverwrite: bool) Tuple[int, Dict[str, Any]][source]

ファイルをコピーする

Parameters:
  • from_path (str) -- コピー元パス

  • to_path (str) -- コピー先パス

  • orverwrite (bool) -- 上書きするかどうか

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_download(current_path: str, img_thumbnail: float = 0.0) Tuple[int, Dict[str, Any]][source]

ファイルをダウンロードする

Parameters:
  • current_path (str) -- ファイルパス

  • img_thumbnail (float, optional) -- サムネイルのサイズ, by default 0.0

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_list(current_path: str, recursive: bool = False) Tuple[int, Dict[str, Any]][source]

ファイルリストを取得する

Parameters:
  • path (str) -- ファイルパス

  • recursive (bool, optional) -- 再帰的に取得するかどうか, by default False

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_mkdir(current_path: str) Tuple[int, Dict[str, Any]][source]

ディレクトリを作成する

Parameters:

current_path (str) -- ディレクトリパス

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_move(from_path: str, to_path: str) Tuple[int, Dict[str, Any]][source]

ファイルを移動する

Parameters:
  • from_path (str) -- 移動元パス

  • to_path (str) -- 移動先パス

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_remove(current_path: str) Tuple[int, Dict[str, Any]][source]

ファイルを削除する

Parameters:

current_path (str) -- ファイルパス

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_rmdir(current_path: str) Tuple[int, Dict[str, Any]][source]

ディレクトリを削除する

Parameters:

current_path (str) -- ディレクトリパス

Returns:

レスポンスコード dict: メッセージ

Return type:

int

file_upload(current_path: str, file_name: str, file_data: bytes, mkdir: bool, orverwrite: bool) Tuple[int, Dict[str, Any]][source]

ファイルをアップロードする

Parameters:
  • current_path (str) -- ファイルパス

  • file_name (str) -- ファイル名

  • file_data (bytes) -- ファイルデータ

  • mkdir (bool) -- ディレクトリを作成するかどうか

  • orverwrite (bool) -- 上書きするかどうか

Returns:

レスポンスコード dict: メッセージ

Return type:

int

cmdbox.app.options module

class cmdbox.app.options.Options(appcls=None, ver=None)[source]

Bases: object

AT_ADMIN = 'admin'
AT_AUTH = 'auth'
AT_EVENT = 'event'
AT_SYSTEM = 'system'
AT_USER = 'user'
AUDITS = ['user', 'admin', 'system', 'auth', 'event']
T_BOOL = 'bool'
T_DATE = 'date'
T_DATETIME = 'datetime'
T_DICT = 'dict'
T_DIR = 'dir'
T_FILE = 'file'
T_FLOAT = 'float'
T_INT = 'int'
T_STR = 'str'
T_TEXT = 'text'
static audit(body: Dict[str, Any] = None, audit_type: str = None, tags: List[str] = None, src: str = None) int[source]

監査ログを書き込む関数を返します。 デコレーターとして使用することができます。

Parameters:
  • body (Dict[str, Any]) -- 監査ログの内容

  • audit_type (str) -- 監査の種類

  • tags (List[str]) -- メッセージのタグ

  • src (str) -- メッセージの発生源

Returns:

レスポンスコード

Return type:

int

audit_exec(*args, body: Dict[str, Any] = None, audit_type: str = None, tags: List[str] = None, src: str = None, title: str = None, user: str = None, **kwargs) None[source]

監査ログを書き込みます。

Parameters:
  • args (Any) -- 呼び出し元で使用している引数

  • body (Dict[str, Any]) -- 監査ログの内容

  • audit_type (str) -- 監査の種類

  • tags (List[str]) -- メッセージのタグ

  • src (str) -- メッセージの発生源

  • title (str) -- メッセージのタイトル

  • user (str) -- メッセージを発生させたユーザー名

  • kwargs (Any) -- 呼び出し元で使用しているキーワード引数

check_agentrule(mode: str, cmd: str, logger: Logger) bool[source]

エージェントが使用してよいコマンドかどうかをチェックします

Parameters:
  • mode (str) -- モード

  • cmd (str) -- コマンド

Returns:

認可されたかどうか

Return type:

bool

static getInstance(appcls=None, ver=None)[source]
get_cmd_attr(mode: str, cmd: str, attr: str) Any[source]

コマンドの属性を取得します。 :param mode: 起動モード :param cmd: コマンド :param attr: 属性

Returns:

属性の値

Return type:

Any

get_cmd_choices(mode: str, cmd: str, webmode: bool = False) List[Dict[str, Any]][source]

コマンドのオプション一覧を取得します。 :param mode: 起動モード :param cmd: コマンド :param webmode: Webモードからの呼び出し. Defaults to False :type webmode: bool, optional

Returns:

オプションの選択肢

Return type:

List[Dict[str, Any]]

get_cmd_keys(mode: str) List[str][source]
get_cmd_opt(mode: str, cmd: str, opt: str, webmode: bool = False) Dict[str, Any][source]

コマンドのオプションを取得します。 :param mode: 起動モード :param cmd: コマンド :param opt: オプション :param webmode: Webモードからの呼び出し. Defaults to False :type webmode: bool, optional

Returns:

オプションの値

Return type:

Dict[str, Any]

get_cmds(mode: str) List[Dict[str, str]][source]

コマンドの選択肢を取得します。 :param mode: 起動モード

Returns:

コマンドの選択肢

Return type:

List[Dict[str, str]]

get_mode_keys() List[str][source]
get_modes() List[Dict[str, str]][source]

起動モードの選択肢を取得します。 :returns: 起動モードの選択肢 :rtype: List[Dict[str, str]]

get_svcmd_feature(svcmd: str) Any[source]

サーバー側のコマンドのフューチャーを取得します。

Parameters:

svcmd -- サーバー側のコマンド

Returns:

フューチャー

Return type:

feature.Feature

init_debugoption()[source]
init_options()[source]
is_features_loaded(ftype: str) bool[source]

指定されたフィーチャータイプが読み込まれているかどうかを返します。

Parameters:

ftype (str) -- フィーチャータイプ

Returns:

読み込まれているかどうか

Return type:

bool

list_options()[source]
load_features_agentrule(logger: Logger)[source]
load_features_aliases_cli(logger: Logger)[source]
load_features_aliases_web(routes: List[Route], logger: Logger)[source]
load_features_args(args_dict: Dict[str, Any])[source]
load_features_audit(logger: Logger)[source]
load_features_file(ftype: str, func, appcls, ver, logger: Logger = None)[source]

フィーチャーファイル(features.yml)を読み込みます。

Parameters:
  • ftype (str) -- フィーチャータイプ。cli又はweb

  • func (Any) -- フィーチャーの処理関数

  • appcls (Any) -- アプリケーションクラス

  • ver (Any) -- バージョンモジュール

  • logger (logging.Logger) -- ロガー

load_svcmd(package_name: str, prefix: str = 'cmdbox_', excludes: list = [], appcls=None, ver=None, logger: Logger = None, isloaded: bool = True)[source]

指定されたパッケージの指定された接頭語を持つモジュールを読み込みます。

Parameters:
  • package_name (str) -- パッケージ名

  • prefix (str) -- 接頭語

  • excludes (list) -- 除外するモジュール名のリスト

  • appcls (Any) -- アプリケーションクラス

  • ver (Any) -- バージョンモジュール

  • logger (logging.Logger) -- ロガー

  • isloaded (bool) -- 読み込み済みかどうか

mk_opt_list(opt: dict, webmode: bool = False) List[str][source]

cmdbox.app.server module

class cmdbox.app.server.Server(data_dir: Path, logger: Logger, redis_host: str = 'localhost', redis_port: int = 6379, redis_password: str = None, svname: str = 'server')[source]

Bases: Filer

list_server() Dict[str, List[Dict[str, Any]]][source]

起動しているサーバーリストを取得する

Returns:

サーバーのリスト

Return type:

Dict[str, List[Dict[str, Any]]]

start_server(retry_count: int = 20, retry_interval: int = 5)[source]

サーバー処理を開始する

terminate_server()[source]

サーバー処理を終了する

cmdbox.app.web module

class cmdbox.app.web.RaiseThread(*args, **kwargs)[source]

Bases: Thread

get_id()[source]
raise_exception()[source]
set_id_and_run()[source]
class cmdbox.app.web.ThreadedUvicorn(logger: Logger, config: Config, guvicorn_config: Dict[str, Any] = None, force_uvicorn: bool = False)[source]

Bases: object

is_alive()[source]
start()[source]
stop()[source]
async wait_for_started()[source]
class cmdbox.app.web.Web(logger: Logger, data: Path, appcls=None, ver=None, redis_host: str = 'localhost', redis_port: int = 6379, redis_password: str = None, svname: str = 'server', client_only: bool = False, doc_root: Path = None, gui_html: str = None, filer_html: str = None, result_html: str = None, users_html: str = None, audit_html: str = None, agent_html: str = None, assets: List[str] = None, signin_html: str = None, signin_file: str = None, gui_mode: bool = False, web_features_packages: List[str] = None, web_features_prefix: List[str] = [])[source]

Bases: object

apikey_add(user: Dict[str, Any]) str[source]

サインインファイルにユーザーのApiKeyを追加する

Parameters:

user (Dict[str, Any]) -- ユーザー情報

Returns:

ApiKey

Return type:

str

apikey_del(user: Dict[str, Any])[source]

サインインファイルのユーザーのApiKeyを削除する

Parameters:

user (Dict[str, Any]) -- ユーザー情報

change_password(user_name: str, password: str, new_password: str, confirm_password: str)[source]

パスワードを変更する

Parameters:
  • user_name (str) -- ユーザー名

  • new_password (str) -- 新しいパスワード

  • confirm_password (str) -- 確認用パスワード

Raises:
  • HTTPException -- パスワードが一致しない場合

  • HTTPException -- ユーザーが存在しない場合

group_add(group: Dict[str, Any])[source]

サインインファイルにグループを追加する

Parameters:

group (Dict[str, Any]) -- グループ情報

group_del(gid: int)[source]

サインインファイルからグループを削除する

Parameters:

gid (int) -- グループID

group_edit(group: Dict[str, Any])[source]

サインインファイルのグループ情報を編集する

Parameters:

group (Dict[str, Any]) -- グループ情報

group_list(name: str = None) List[Dict[str, Any]][source]

サインインファイルのグループ一覧を取得する

Parameters:

name (str, optional) -- グループ名. Defaults to None.

Returns:

グループ一覧

Return type:

List[Dict[str, Any]]

init_webfeatures(app: FastAPI)[source]
start(allow_host: str = '0.0.0.0', listen_port: int = 8081, ssl_listen_port: int = 8443, ssl_cert: Path = None, ssl_key: Path = None, ssl_keypass: str = None, ssl_ca_certs: Path = None, session_domain: str = None, session_path: str = '/', session_secure: bool = False, session_timeout: int = 900, outputs_key: List[str] = [], guvicorn_workers: int = -1, guvicorn_timeout: int = 30, agent_runner=None, mcp=None)[source]

Webサーバを起動する

Parameters:
  • allow_host (str, optional) -- 許可ホスト. Defaults to "

  • listen_port (int, optional) -- リスンポート. Defaults to 8081.

  • ssl_listen_port (int, optional) -- SSLリスンポート. Defaults to 8443.

  • ssl_cert (Path, optional) -- SSL証明書ファイル. Defaults to None.

  • ssl_key (Path, optional) -- SSL秘密鍵ファイル. Defaults to None.

  • ssl_keypass (str, optional) -- SSL秘密鍵パスワード. Defaults to None.

  • ssl_ca_certs (Path, optional) -- SSL CA証明書ファイル. Defaults to None.

  • session_domain (str, optional) -- セッションドメイン. Defaults to None.

  • session_path (str, optional) -- セッションパス. Defaults to '/'.

  • session_secure (bool, optional) -- セッションセキュア. Defaults to False.

  • session_timeout (int, optional) -- セッションタイムアウト. Defaults to 900.

  • outputs_key (list, optional) -- 出力キー. Defaults to [].

  • guvicorn_workers (int, optional) -- Gunicornワーカー数. Defaults to -1.

  • guvicorn_timeout (int, optional) -- Gunicornタイムアウト. Defaults to 30.

  • agent_runner (Runner, optional) -- エージェントランナー. Defaults to None.

  • mcp (MCP, optional) -- MCP. Defaults to None.

stop()[source]

Webサーバを停止する

user_add(user: Dict[str, Any])[source]

サインインファイルにユーザーを追加する

Parameters:

user (Dict[str, Any]) -- ユーザー情報

user_data(req: Request, uid: str, user_name: str, categoly: str, key: str = None, val: Any = None, delkey: bool = False) Any[source]

ユーザーデータを取得または設定する

Parameters:
  • req (Request) -- リクエスト

  • uid (str) -- ユーザーID

  • user_name (str) -- ユーザー名

  • categoly (str) -- カテゴリ

  • key (str, optional) -- キー. Defaults to None.

  • val (Any, optional) -- 値. Defaults to None.

  • delkey (bool, optional) -- キー削除. Defaults to False.

Returns:

値 or カテゴリのデータ

Return type:

Any

user_del(uid: int)[source]

サインインファイルからユーザーを削除する

Parameters:

uid (int) -- ユーザーID

user_edit(user: Dict[str, Any])[source]

サインインファイルのユーザー情報を編集する

Parameters:

user (Dict[str, Any]) -- ユーザー情報

user_list(name: str = None) List[Dict[str, Any]][source]

サインインファイルのユーザー一覧を取得する

Parameters:

name (str, optional) -- ユーザー名. Defaults to None.

Returns:

ユーザー一覧

Return type:

List[Dict[str, Any]]

Module contents