pingu_98/MrDraw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

First steps towards API based CLI commands - "octoprint client"

Browse source
This commit is contained in:
Gina Häußge 2015-11-04 18:54:51 +01:00
parent 4fcb853ef2
commit 2214b2ac42
3 changed files with 247 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/octoprint/cli/__init__.py
View file

@ -119,9 +119,10 @@ legacy_options = bulk_options([
from .server import server_commands
from .plugins import plugin_commands
from .dev import dev_commands
from .client import client_commands
@click.group(name="octoprint", invoke_without_command=True, cls=click.CommandCollection,
sources=[server_commands, plugin_commands, dev_commands])
sources=[server_commands, plugin_commands, dev_commands, client_commands])
@standard_options()
@legacy_options
@click.version_option(version=octoprint.__version__)

172
src/octoprint/cli/client.py Normal file
View file

@ -0,0 +1,172 @@
# coding=utf-8
from __future__ import absolute_import
__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
import click
import json
import octoprint_client
from octoprint.cli import pass_octoprint_ctx, bulk_options, standard_options
from octoprint import init_settings
class JsonStringParamType(click.ParamType):
name = "json"
def convert(self, value, param, ctx):
try:
return json.loads(value)
except:
self.fail("%s is not a valid json string" % value, param, ctx)
@click.group()
def client_commands():
pass
@client_commands.group("client", context_settings=dict(ignore_unknown_options=True))
@click.option("--host", "-h", type=click.STRING)
@click.option("--port", "-p", type=click.INT)
@click.option("--httpuser", type=click.STRING)
@click.option("--httppass", type=click.STRING)
@click.option("--https", is_flag=True)
@click.option("--prefix", type=click.STRING)
@pass_octoprint_ctx
def client(obj, host, port, httpuser, httppass, https, prefix):
"""Basic API client."""
obj.settings = init_settings(obj.basedir, obj.configfile)
settings_host = obj.settings.get(["server", "host"])
settings_port = obj.settings.getInt(["server", "port"])
settings_apikey = obj.settings.get(["api", "key"])
octoprint_client.apikey = settings_apikey
octoprint_client.baseurl = build_base_url(https=https,
httpuser=httpuser,
httppass=httppass,
host=host or settings_host if settings_host != "0.0.0.0" else "127.0.0.1",
port=port or settings_port,
prefix=prefix)
def build_base_url(https=False, httpuser=None, httppass=None, host=None, port=None, prefix=None):
protocol = "https" if https else "http"
httpauth = "{}:{}@".format(httpuser, httppass) if httpuser and httppass else ""
host = host if host else "127.0.0.1"
port = ":{}".format(port) if port else ":5000"
prefix = prefix if prefix else ""
return "{}://{}{}{}{}".format(protocol, httpauth, host, port, prefix)
def build_url(obj, path):
return "{}/{}".format(build_base_url(obj), path)
def log_response(response, status_code=True, body=True, headers=False):
if status_code:
click.echo("Status Code: {}".format(response.status_code))
if headers:
for header, value in response.headers.items():
click.echo("{}: {}".format(header, value))
click.echo()
if body:
click.echo(response.text)
@client.command("get")
@click.argument("path")
def get(path):
"""Performs a GET request against the specified server path."""
r = octoprint_client.get(path)
log_response(r)
@client.command("post_json")
@click.argument("path")
@click.argument("data", type=JsonStringParamType())
def post_json(path, data):
"""POSTs JSON data to the specified server path."""
r = octoprint_client.post_json(path, data)
log_response(r)
@client.command("patch_json")
@click.argument("path")
@click.argument("data", type=JsonStringParamType())
def patch_json(path, data):
"""PATCHes JSON data to the specified server path."""
r = octoprint_client.patch(path, data, encoding="json")
log_response(r)
@client.command("post_from_file")
@click.argument("path")
@click.argument("file_path", type=click.Path(exists=True, dir_okay=False, resolve_path=True))
@click.option("--json", is_flag=True)
@click.option("--yaml", is_flag=True)
def post_from_file(path, file_path, json_flag, yaml_flag):
"""POSTs JSON data to the specified server path."""
if json_flag or yaml_flag:
if json_flag:
with open(file_path, "rb") as fp:
data = json.load(fp)
else:
import yaml
with open(file_path, "rb") as fp:
data = yaml.safe_load(fp)
r = octoprint_client.post_json(path, data)
else:
with open(file_path, "rb") as fp:
data = fp.read()
r = octoprint_client.post(path, data)
log_response(r)
@client.command("command")
@click.argument("path")
@click.argument("command")
@click.option("--str", "-s", "str_params", multiple=True, nargs=2, type=click.Tuple([unicode, unicode]))
@click.option("--int", "-i", "int_params", multiple=True, nargs=2, type=click.Tuple([unicode, int]))
@click.option("--float", "-f", "float_params", multiple=True, nargs=2, type=click.Tuple([unicode, float]))
@click.option("--bool", "-b", "bool_params", multiple=True, nargs=2, type=click.Tuple([unicode, bool]))
def command(path, command, str_params, int_params, float_params, bool_params):
"""Sends a JSON command to the specified server path."""
data = dict()
params = str_params + int_params + float_params + bool_params
for param in params:
data[param[0]] = param[1]
r = octoprint_client.post_command(path, command, parameters=data)
log_response(r, body=False)
@client.command("upload")
@click.argument("path")
@click.argument("file_path", type=click.Path(exists=True, dir_okay=False, resolve_path=True))
@click.option("--parameter", "-P", "params", multiple=True, nargs=2, type=click.Tuple([unicode, unicode]))
@click.option("--file-name", type=click.STRING)
@click.option("--content-type", type=click.STRING)
def upload(path, file_path, params, file_name, content_type):
"""Uploads the specified file to the specified server path."""
data = dict()
for param in params:
data[param[0]] = param[1]
r = octoprint_client.upload(path, file_path, parameters=data, file_name=file_name, content_type=content_type)
log_response(r)
@client.command("delete")
@click.argument("path")
def delete(path):
"""Sends a DELETE request to the specified server path."""
r = octoprint_client.delete(path)
log_response(r)

73
src/octoprint_client/__init__.py Normal file
View file

@ -0,0 +1,73 @@
# coding=utf-8
from __future__ import absolute_import
__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
import requests
apikey = None
baseurl = None
def prepare_request(method=None, path=None):
url = None
if baseurl:
while path.startswith("/"):
path = path[1:]
url = baseurl + "/" + path
return requests.Request(method=method, url=url, headers={"X-Api-Key": apikey}).prepare()
def request(method, path, data=None, files=None, encoding=None):
s = requests.Session()
request = prepare_request(method, path)
if data or files:
if encoding == "json":
request.prepare_body(None, None, json=data)
else:
request.prepare_body(data, files=files)
response = s.send(request)
return response
def get(path):
return request("GET", path)
def post(path, data, encoding=None):
return request("POST", path, data=data, encoding=encoding)
def post_json(path, data):
return post(path, data, encoding="json")
def post_command(path, command, parameters=None):
data = dict(command=command)
if parameters:
data.update(parameters)
return post_json(path, data)
def upload(path, file_path, parameters=None, file_name=None, content_type=None):
import os
if not os.path.isfile(file_path):
raise ValueError("{} cannot be uploaded since it is not a file".format(file_path))
if file_name is None:
file_name = os.path.basename(file_path)
with open(file_path, "rb") as fp:
if content_type:
files = dict(file=(file_name, fp, content_type))
else:
files = dict(file=(file_name, fp))
response = request("POST", path, data=parameters, files=files)
return response
def delete(path):
return request("DELETE", path)
def patch(path, data, encoding=None):
return request("PATCH", path, data=data, encoding=encoding)
def put(path, data, encoding=None):
return request("PUT", path, data=data, encoding=encoding)