プロジェクト概要
プロジェクト概要
プロジェクトのURL: https://github.com/milvus-io/milvus_cli
準備Python3.8
, Click 8.0.x
グループコマンド
コマンドの作成
import click
from utils import PyOrm
@click.group(no_args_is_help=False, add_help_option=False, invoke_without_command=True)
@click.pass_context
def cli(ctx):
"""Milvus CLI"""
ctx.obj = PyOrm() # PyOrm is a util class which wraps the milvus python SDK. You can pass any class instance here. Any command function passed by @click.obj can call it.
if __name__ == '__main__':
cli()
上のコードのように、@click.group()
を使って、cli
をエントリー・ポイントとしてコマンド・グループを作成する。プロンプトCLIを実装するには、エントリーのヘルプ・メッセージを無効にする必要があるので、no_args_is_help=False
、add_help_option=False
、invoke_without_command=True
を追加する。また、cli
をターミナルに入力するだけでは、何も表示されない。
さらに、@click.pass_context
を使って、このグループにコンテキストを渡し、さらに使用できるようにする。
コマンドグループのサブコマンドの作成
最初のサブコマンドhelp
をcli
の下に追加する:
# Print the help message of specified command.
def print_help_msg(command):
with click.Context(command) as ctx:
click.echo(command.get_help(ctx))
# Use @cli.command() to create a sub command of cli.
@cli.command()
def help():
"""Show help messages."""
# Print help message of cli.
click.echo(print_help_msg(cli))
これで、ターミナルでcli help
:
$ python milvus_cli/scripts/milvus_cli.py help
コマンドグループのサブグループの作成
cli help
のようなサブコマンドだけでなく、cli list collection
、cli list partition
、cli list indexes
のようなサブグループコマンドも必要だ。
まず、サブグループコマンドlist
を作成します。ここでは、@cli.group
の最初のパラメータをコマンド名として渡すことで、関数名の重複を減らすことができます。
ここでは、@click.group
の代わりに@cli.group()
を使用し、オリジングループのサブグループを作成します。
このサブグループのサブコマンドにcontext.obj
を渡すために@click.pass_obj
を使用します。
@cli.group('list', no_args_is_help=False)
@click.pass_obj
def listDetails(obj):
"""List collections, partitions and indexes."""
pass
そして、@listDetails.command()
(@cli.command()
ではありません)を使って、このサブグループにいくつかのサブコマンドを追加します。これはほんの一例です。実装については後ほど説明しますので、無視してかまいません。
@listDetails.command()
@click.option('--timeout', 'timeout', help="[Optional] - An optional duration of time in seconds to allow for the RPC. When timeout is set to None, client waits until server response or error occur.", default=None)
@click.option('--show-loaded', 'showLoaded', help="[Optional] - Only show loaded collections.", default=False)
@click.pass_obj
def collections(obj, timeout, showLoaded):
"""List all collections."""
try:
obj.checkConnection()
click.echo(obj.listCollections(timeout, showLoaded))
except Exception as e:
click.echo(message=e, err=True)
@listDetails.command()
@click.option('-c', '--collection', 'collection', help='The name of collection.', default='')
@click.pass_obj
def partitions(obj, collection):
"""List all partitions of the specified collection."""
try:
obj.checkConnection()
validateParamsByCustomFunc(
obj.getTargetCollection, 'Collection Name Error!', collection)
click.echo(obj.listPartitions(collection))
except Exception as e:
click.echo(message=e, err=True)
これらがすべて完了すると、次のようなmiltigroupコマンドができあがります:
イメージ
カスタムコマンド
オプションの追加
cli --test-option value
のように、コマンドにオプションを追加することができます。
ここでは、milvusに接続するためのアドレスを指定するために、alias
、host
、port
の3つのオプションを追加します。
最初の2つのパラメータは短いオプション名と完全なオプション名を定義し、3番目のパラメータは変数名を定義し、help
パラメータは短いヘルプメッセージを指定し、default
パラメータはデフォルト値を指定し、type
は値のタイプを指定します。
そして、すべてのオプションの値は、定義順に関数に渡されます。
@cli.command(no_args_is_help=False)
@click.option('-a', '--alias', 'alias', help="Milvus link alias name, default is `default`.", default='default', type=str)
@click.option('-h', '--host', 'host', help="Host name, default is `127.0.0.1`.", default='127.0.0.1', type=str)
@click.option('-p', '--port', 'port', help="Port, default is `19530`.", default=19530, type=int)
@click.pass_obj
def connect(obj, alias, host, port):
pass
フラグ・オプションの追加
上記ではオプションを使って値を渡しましたが、ブーリアン値としてフラグが必要な場合もあります。
以下の例のように、autoId
はフラグオプションで、関数にデータを渡さないので、cli create collection -c c_name -p p_name -a
のように使うことができます。
@createDetails.command('collection')
@click.option('-c', '--collection-name', 'collectionName', help='Collection name to be created.', default='')
@click.option('-p', '--schema-primary-field', 'primaryField', help='Primary field name.', default='')
@click.option('-a', '--schema-auto-id', 'autoId', help='Enable auto id.', default=False, is_flag=True)
@click.pass_obj
def createCollection(obj, collectionName, primaryField, autoId, description, fields):
pass
引数の追加
このプロジェクトでは、すべての引数の使い方をオプションの使い方に置き換えています。しかし、ここではまだ引数の使い方を紹介します。オプションとは異なり、引数はcli COMMAND [OPTIONS] ARGUEMENTS
のように使用します。上の例をargumentsの使い方に変換すると、このようになります:
@createDetails.command('collection')
@click.argument('collectionName')
@click.option('-p', '--schema-primary-field', 'primaryField', help='Primary field name.', default='')
@click.option('-a', '--schema-auto-id', 'autoId', help='Enable auto id.', default=False, is_flag=True)
@click.pass_obj
def createCollection(obj, collectionName, primaryField, autoId, description, fields):
pass
そうすると、使い方はcli create collection c_name -p p_name -a
のようになります。
完全なヘルプメッセージを追加する
上で短いヘルプメッセージを定義したように、関数の中で完全なヘルプメッセージを定義することができます:
@cli.command(no_args_is_help=False)
@click.option('-a', '--alias', 'alias', help="Milvus link alias name, default is `default`.", default='default', type=str)
@click.option('-h', '--host', 'host', help="Host name, default is `127.0.0.1`.", default='127.0.0.1', type=str)
@click.option('-p', '--port', 'port', help="Port, default is `19530`.", default=19530, type=int)
@click.pass_obj
def connect(obj, alias, host, port):
"""
Connect to Milvus.
Example:
milvus_cli > connect -h 127.0.0.1 -p 19530 -a default
"""
try:
obj.connect(alias, host, port)
except Exception as e:
click.echo(message=e, err=True)
else:
click.echo("Connect Milvus successfully!")
click.echo(obj.showConnection(alias))
関数内の最初のブロックは、cli connect --help
を入力した後に表示されるヘルプメッセージです。
milvus_cli > connect --help
Usage: milvus_cli.py connect [OPTIONS]
Connect to Milvus.
Example:
milvus_cli > connect -h 127.0.0.1 -p 19530 -a default
Options:
-a, --alias TEXT Milvus link alias name, default is `default`.
-h, --host TEXT Host name, default is `127.0.0.1`.
-p, --port INTEGER Port, default is `19530`.
--help Show this message and exit.
確認の追加
時には、ユーザーに何かのアクションを確認してもらう必要があります。click.confirm
、一時停止してユーザーに確認を求めることができます:
@deleteSth.command('collection')
@click.option('-c', '--collection', 'collectionName', help='The name of collection to be deleted.', default='')
@click.option('-t', '--timeout', 'timeout', help='An optional duration of time in seconds to allow for the RPC. If timeout is set to None, the client keeps waiting until the server responds or an error occurs.', default=None, type=int)
@click.pass_obj
def deleteCollection(obj, collectionName, timeout):
"""
Drops the collection together with its index files.
Example:
milvus_cli > delete collection -c car
"""
click.echo(
"Warning!\nYou are trying to delete the collection with data. This action cannot be undone!\n")
if not click.confirm('Do you want to continue?'):
return
pass
上記の例のように、確認の会話はAborted!ant to continue? [y/N]:
のように表示されます。
プロンプトの追加
プロンプトを実装するには、click.prompt
を追加するだけです。
@cli.command()
@click.pass_obj
def query(obj):
"""
Query with a set of criteria, and results in a list of records that match the query exactly.
"""
collectionName = click.prompt(
'Collection name', type=click.Choice(obj._list_collection_names()))
expr = click.prompt('The query expression(field_name in [x,y])')
partitionNames = click.prompt(
f'The names of partitions to search(split by "," if multiple) {obj._list_partition_names(collectionName)}', default='')
outputFields = click.prompt(
f'Fields to return(split by "," if multiple) {obj._list_field_names(collectionName)}', default='')
timeout = click.prompt('timeout', default='')
pass
プロンプトは、click.prompt
が表示されるたびに表示されます。いくつかのプロンプトを連続して使用することで、連続した会話に見えるようにします。こうすることで、ユーザーは私たちが望む順番でデータを入力するようになります。この場合、ユーザーにまずコレクションを選んでもらい、そのコレクションの下にあるすべてのパーティションを取得し、ユーザーにそれを見せて選んでもらう必要があります。
選択肢の追加
ユーザーが限られた範囲/種類の値だけを入力したい場合があります。その場合、type=click.Choice([<any>])
をclick.prompt
やclick.options
などに追加することができます。
例えば
collectionName = click.prompt(
'Collection name', type=click.Choice(['collection_1', 'collection_2']))
この場合、ユーザはcollection_1
かcollection_2
しか入力できません。他の入力があるとエラーになります。
クリア画面の追加
click.clear()
。
@cli.command()
def clear():
"""Clear screen."""
click.clear()
追加のヒント
- デフォルト値は
None
ですので、デフォルト値をNone
に指定しても意味がありません。また、デフォルトのNone
では、click.prompt
が表示され続けるので、値を空にしておくと飛び越えることができます。
プロンプトCLIの実装
なぜプロンプトCLIなのか
データベースを操作するためには、インスタンスへの継続的な接続が必要です。オリジンのコマンドラインモードを使用すると、コマンドを実行するたびに接続が切断されてしまいます。また、CLIを使用する際にいくつかのデータを保存し、終了後にそれらを消去したい。
実装
- ユーザーの入力を継続的にリッスンするには、
while True
。
def runCliPrompt():
while True:
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
if __name__ == '__main__':
runCliPrompt()
input
のみを使用すると、up
、down
、left
、right
矢印キー、tab
キー、その他いくつかのキーが自動的にAcsii文字列に変換される。また、履歴コマンドはセッションから読み取ることができない。そこで、runCliPrompt
関数にreadline
を追加する。
def runCliPrompt():
while True:
import readline
readline.set_completer_delims(' \t\n;')
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
quit
CLIを追加。
@cli.command('exit')
def quitapp():
"""Exit the CLI."""
global quitapp
quitapp = True
quitapp = False # global flag
def runCliPrompt():
while not quitapp:
import readline
readline.set_completer_delims(' \t\n;')
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
ctrl C
で終了するときにKeyboardInterrupt
のエラーをキャッチする。
def runCliPrompt():
try:
while not quitapp:
import readline
readline.set_completer_delims(' \t\n;')
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
except KeyboardInterrupt:
sys.exit(0)
- これでCLIは以下のようになった:
milvus_cli >
milvus_cli > connect
+-------+-----------+
| Host | 127.0.0.1 |
| Port | 19530 |
| Alias | default |
+-------+-----------+
milvus_cli > help
Usage: [OPTIONS] COMMAND [ARGS]...
Milvus CLI
Commands:
clear Clear screen.
connect Connect to Milvus.
create Create collection, partition and index.
delete Delete specified collection, partition and index.
describe Describe collection or partition.
exit Exit the CLI.
help Show help messages.
import Import data from csv file with headers and insert into target...
list List collections, partitions and indexes.
load Load specified collection.
query Query with a set of criteria, and results in a list of...
release Release specified collection.
search Conducts a vector similarity search with an optional boolean...
show Show connection, loading_progress and index_progress.
version Get Milvus CLI version.
milvus_cli > exit
オートコンプリートを手動で実装する
clickのシェル・オートコンプリートとは異なり、このプロジェクトではコマンドラインをラップし、ループを使ってユーザーの入力を取得し、プロンプト・コマンドラインを実装する。そのため、readline
にコンプレッターをバインドする必要がある。
class Completer(object):
RE_SPACE = re.compile('.*\s+$', re.M)
CMDS_DICT = {
'clear': [],
'connect': [],
'create': ['collection', 'partition', 'index'],
'delete': ['collection', 'partition', 'index'],
'describe': ['collection', 'partition'],
'exit': [],
'help': [],
'import': [],
'list': ['collections', 'partitions', 'indexes'],
'load': [],
'query': [],
'release': [],
'search': [],
'show': ['connection', 'index_progress', 'loading_progress'],
'version': [],
}
def __init__(self) -> None:
super().__init__()
self.COMMANDS = list(self.CMDS_DICT.keys())
self.createCompleteFuncs(self.CMDS_DICT)
def createCompleteFuncs(self, cmdDict):
for cmd in cmdDict:
sub_cmds = cmdDict[cmd]
complete_example = self.makeComplete(cmd, sub_cmds)
setattr(self, 'complete_%s' % cmd, complete_example)
def makeComplete(self, cmd, sub_cmds):
def f_complete(args):
f"Completions for the {cmd} command."
if not args:
return self._complete_path('.')
if len(args) <= 1 and not cmd == 'import':
return self._complete_2nd_level(sub_cmds, args[-1])
return self._complete_path(args[-1])
return f_complete
def _listdir(self, root):
"List directory 'root' appending the path separator to subdirs."
res = []
for name in os.listdir(root):
path = os.path.join(root, name)
if os.path.isdir(path):
name += os.sep
res.append(name)
return res
def _complete_path(self, path=None):
"Perform completion of filesystem path."
if not path:
return self._listdir('.')
dirname, rest = os.path.split(path)
tmp = dirname if dirname else '.'
res = [os.path.join(dirname, p)
for p in self._listdir(tmp) if p.startswith(rest)]
# more than one match, or single match which does not exist (typo)
if len(res) > 1 or not os.path.exists(path):
return res
# resolved to a single directory, so return list of files below it
if os.path.isdir(path):
return [os.path.join(path, p) for p in self._listdir(path)]
# exact file match terminates this completion
return [path + ' ']
def _complete_2nd_level(self, SUB_COMMANDS=[], cmd=None):
if not cmd:
return [c + ' ' for c in SUB_COMMANDS]
res = [c for c in SUB_COMMANDS if c.startswith(cmd)]
if len(res) > 1 or not (cmd in SUB_COMMANDS):
return res
return [cmd + ' ']
def complete(self, text, state):
"Generic readline completion entry point."
buffer = readline.get_line_buffer()
line = readline.get_line_buffer().split()
# show all commands
if not line:
return [c + ' ' for c in self.COMMANDS][state]
# account for last argument ending in a space
if self.RE_SPACE.match(buffer):
line.append('')
# resolve command to the implementation function
cmd = line[0].strip()
if cmd in self.COMMANDS:
impl = getattr(self, 'complete_%s' % cmd)
args = line[1:]
if args:
return (impl(args) + [None])[state]
return [cmd + ' '][state]
results = [
c + ' ' for c in self.COMMANDS if c.startswith(cmd)] + [None]
return results[state]
Completer
を定義したら、readlineとバインドすることができる:
comp = Completer()
def runCliPrompt():
try:
while not quitapp:
import readline
readline.set_completer_delims(' \t\n;')
readline.parse_and_bind("tab: complete")
readline.set_completer(comp.complete)
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
except KeyboardInterrupt:
sys.exit(0)
ワンタイムオプションの追加
プロンプト・コマンド・ラインでは、バージョンなどの情報を得るためにスクリプトを完全に実行したくないことがあります。良い例がPython
です。ターミナルでpython
と入力すると、promtp コマンドラインが表示されますが、バージョンメッセージが返されるだけで、python -V
と入力してもプロンプトスクリプトには入りません。そこで、sys.args
を使って実装することができる。
def runCliPrompt():
args = sys.argv
if args and (args[-1] == '--version'):
print(f"Milvus Cli v{getPackageVersion()}")
return
try:
while not quitapp:
import readline
readline.set_completer_delims(' \t\n;')
readline.parse_and_bind("tab: complete")
readline.set_completer(comp.complete)
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
except KeyboardInterrupt:
sys.exit(0)
if __name__ == '__main__':
runCliPrompt()
最初にCLIスクリプトを実行するとき、ループの前にsys.args
。最後の引数が--version
の場合、ループに入ることなくパッケージのバージョンを返します。
これは、コードをパッケージとしてビルドした後に役立ちます。プロンプトCLIにジャンプするにはmilvus_cli
、バージョンだけを取得するにはmilvus_cli --version
。
ビルドとリリース
最後に、パッケージをビルドし、PYPIでリリースします。ユーザーはpip install <package name>
。
テスト用にローカルにインストールする
PYPIにパッケージを公開する前に、ローカルにインストールしてテストしたい場合があります。
この場合、cd
をパッケージディレクトリにコピーし、pip install -e .
を実行します(.
を忘れないでください)。
パッケージファイルの作成
参照:https://packaging.python.org/tutorials/packaging-projects/
パッケージの構造は以下のようになります:
package_example/
├── LICENSE
├── README.md
├── setup.py
├── src/
│ ├── __init__.py
│ ├── main.py
│ └── scripts/
│ ├── __init__.py
│ └── example.py
└── tests/
パッケージ・ディレクトリの作成
Milvus_cli
ディレクトリを以下のような構造で作成する:
Milvus_cli/
├── LICENSE
├── README.md
├── setup.py
├── milvus_cli/
│ ├── __init__.py
│ ├── main.py
│ ├── utils.py
│ └── scripts/
│ ├── __init__.py
│ └── milvus_cli.py
└── dist/
エントリーコードを書く
スクリプトのエントリーはMilvus_cli/milvus_cli/scripts
、Milvus_cli/milvus_cli/scripts/milvus_cli.py
のようにする:
import sys
import os
import click
from utils import PyOrm, Completer
pass_context = click.make_pass_decorator(PyOrm, ensure=True)
@click.group(no_args_is_help=False, add_help_option=False, invoke_without_command=True)
@click.pass_context
def cli(ctx):
"""Milvus CLI"""
ctx.obj = PyOrm()
"""
...
Here your code.
...
"""
@cli.command('exit')
def quitapp():
"""Exit the CLI."""
global quitapp
quitapp = True
quitapp = False # global flag
comp = Completer()
def runCliPrompt():
args = sys.argv
if args and (args[-1] == '--version'):
print(f"Milvus Cli v{getPackageVersion()}")
return
try:
while not quitapp:
import readline
readline.set_completer_delims(' \t\n;')
readline.parse_and_bind("tab: complete")
readline.set_completer(comp.complete)
astr = input('milvus_cli > ')
try:
cli(astr.split())
except SystemExit:
# trap argparse error message
# print('error', SystemExit)
continue
except Exception as e:
click.echo(
message=f"Error occurred!\n{str(e)}", err=True)
except KeyboardInterrupt:
sys.exit(0)
if __name__ == '__main__':
runCliPrompt()
を編集する。setup.py
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name='milvus_cli',
version='0.1.6',
author='Milvus Team',
author_email='milvus-team@zilliz.com',
url='https://github.com/milvus-io/milvus_cli',
description='CLI for Milvus',
long_description=long_description,
long_description_content_type='text/markdown',
license='Apache-2.0',
packages=find_packages(),
include_package_data=True,
install_requires=[
'Click==8.0.1',
'pymilvus==2.0.0rc5',
'tabulate==0.8.9'
],
entry_points={
'console_scripts': [
'milvus_cli = milvus_cli.scripts.milvus_cli:runCliPrompt',
],
},
python_requires='>=3.8'
)
ここでいくつかのヒントを:
README.md
の内容をパッケージの長い説明として使う。- すべての依存関係を
install_requires
に追加する。 entry_points
を指定する。この場合、milvus_cli
をconsole_scripts
の子として設定し、このパッケージをインストールした後、milvus_cli
を直接コマンドとして入力できるようにする。また、milvus_cli
のエントリー・ポイントはmilvus_cli/scripts/milvus_cli.py
のrunCliPrompt
関数である。
ビルド
build
パッケージをアップグレードする:python3 -m pip install --upgrade build
buildを実行する:
python -m build --sdist --wheel --outdir dist/ .
dist/
ディレクトリの下に2つのファイルが生成される:
dist/
example_package_YOUR_USERNAME_HERE-0.0.1-py3-none-any.whl
example_package_YOUR_USERNAME_HERE-0.0.1.tar.gz
リリースの公開
参照: https://packaging.python.org/tutorials/packaging-projects/#uploading-the-distribution-archives
twine
パッケージをアップグレードする:python3 -m pip install --upgrade twine
PYPI
test envにアップロードする:python3 -m twine upload --repository testpypi dist/*
PYPI
にアップロードする:python3 -m twine upload dist/*
GithubワークフローによるCI/CD
参考:https://packaging.python.org/guides/publishing-package-distribution-releases-using-github-actions-ci-cd-workflows/
パッケージをビルドして、githubのリリースとPYPIにアップロードする。
(何らかの理由で、PYPIをテストするためにリリースを公開するだけのワークフローが欲しい。)
# This is a basic workflow to help you get started with Actions
name: Update the release's assets after it published
# Controls when the workflow will run
on:
release:
# The workflow will run after release published
types: [published]
# A workflow run is made up of one or more jobs that can run sequentially or in parallel
jobs:
# This workflow contains a single job called "build"
build:
# The type of runner that the job will run on
runs-on: ubuntu-latest
# Steps represent a sequence of tasks that will be executed as part of the job
steps:
# Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
with:
python-version: '3.8'
architecture: 'x64'
- name: Install pypa/build
run: >-
python -m
pip install
build
--user
- name: Clean dist/
run: |
sudo rm -fr dist/*
- name: Build a binary wheel and a source tarball
run: >-
python -m
build
--sdist
--wheel
--outdir dist/
.
# Update target github release's assets
- name: Update assets
uses: softprops/action-gh-release@v1
if: startsWith(github.ref, 'refs/tags/')
with:
files: ./dist/*
- name: Publish distribution 📦 to Test PyPI
if: contains(github.ref, 'beta') && startsWith(github.ref, 'refs/tags')
uses: pypa/gh-action-pypi-publish@release/v1
with:
user: __token__
password: ${{ secrets.TEST_PYPI_API_TOKEN }}
repository_url: https://test.pypi.org/legacy/
packages_dir: dist/
verify_metadata: false
Milvusについてもっと知る
Milvusは、人工知能やベクトル類似検索の膨大なアプリケーションを動かすことができる強力なツールです。プロジェクトの詳細については、以下のリソースをご覧ください:
Try Managed Milvus for Free
Zilliz Cloud is hassle-free, powered by Milvus and 10x faster.
Get StartedLike the article? Spread the word