Changeset - 6d0c2e63f225
[Not reviewed]
default
0 2 6
Dennis Fink - 2 years ago 2022-02-24 21:02:53
dennis.fink@c3l.lu
Implemented first version
8 files changed with 447 insertions and 1 deletions:
0 comments (0 inline, 0 general)
poetry.lock
Show inline comments
 
@@ -58,6 +58,17 @@ optional = false
 
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
 

	
 
[[package]]
 
name = "commonmark"
 
version = "0.9.1"
 
description = "Python parser for the CommonMark Markdown spec"
 
category = "main"
 
optional = false
 
python-versions = "*"
 

	
 
[package.extras]
 
test = ["flake8 (==3.7.8)", "hypothesis (==3.55.3)"]
 

	
 
[[package]]
 
name = "idna"
 
version = "3.3"
 
description = "Internationalized Domain Names in Applications (IDNA)"
 
@@ -125,6 +136,14 @@ docs = ["Sphinx (>=4)", "furo (>=2021.7.
 
test = ["appdirs (==1.4.4)", "pytest (>=6)", "pytest-cov (>=2.7)", "pytest-mock (>=3.6)"]
 

	
 
[[package]]
 
name = "pygments"
 
version = "2.11.2"
 
description = "Pygments is a syntax highlighting package written in Python."
 
category = "main"
 
optional = false
 
python-versions = ">=3.5"
 

	
 
[[package]]
 
name = "requests"
 
version = "2.27.1"
 
description = "Python HTTP for Humans."
 
@@ -143,6 +162,22 @@ socks = ["PySocks (>=1.5.6,!=1.5.7)", "w
 
use_chardet_on_py3 = ["chardet (>=3.0.2,<5)"]
 

	
 
[[package]]
 
name = "rich"
 
version = "11.2.0"
 
description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal"
 
category = "main"
 
optional = false
 
python-versions = ">=3.6.2,<4.0.0"
 

	
 
[package.dependencies]
 
colorama = ">=0.4.0,<0.5.0"
 
commonmark = ">=0.9.0,<0.10.0"
 
pygments = ">=2.6.0,<3.0.0"
 

	
 
[package.extras]
 
jupyter = ["ipywidgets (>=7.5.1,<8.0.0)"]
 

	
 
[[package]]
 
name = "tomli"
 
version = "2.0.0"
 
description = "A lil' TOML parser"
 
@@ -174,7 +209,7 @@ socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0)
 
[metadata]
 
lock-version = "1.1"
 
python-versions = "^3.10"
 
content-hash = "aa2b2339dc49cd0a2f5dbdf90b4e946bd890b422351f4d9f58bd47eb900a0edb"
 
content-hash = "14f945aef88b635fa5d4cc1f329572834551d8fb4fe0c7cbbbcfa419bbec84a7"
 

	
 
[metadata.files]
 
black = [
 
@@ -218,6 +253,10 @@ colorama = [
 
    {file = "colorama-0.4.4-py2.py3-none-any.whl", hash = "sha256:9f47eda37229f68eee03b24b9748937c7dc3868f906e8ba69fbcbdd3bc5dc3e2"},
 
    {file = "colorama-0.4.4.tar.gz", hash = "sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b"},
 
]
 
commonmark = [
 
    {file = "commonmark-0.9.1-py2.py3-none-any.whl", hash = "sha256:da2f38c92590f83de410ba1a3cbceafbc74fee9def35f9251ba9a971d6d66fd9"},
 
    {file = "commonmark-0.9.1.tar.gz", hash = "sha256:452f9dc859be7f06631ddcb328b6919c67984aca654e5fefb3914d54691aed60"},
 
]
 
idna = [
 
    {file = "idna-3.3-py3-none-any.whl", hash = "sha256:84d9dd047ffa80596e0f246e2eab0b391788b0503584e8945f2368256d2735ff"},
 
    {file = "idna-3.3.tar.gz", hash = "sha256:9d643ff0a55b762d5cdb124b8eaa99c66322e2157b69160bc32796e824360e6d"},
 
@@ -260,10 +299,18 @@ platformdirs = [
 
    {file = "platformdirs-2.4.1-py3-none-any.whl", hash = "sha256:1d7385c7db91728b83efd0ca99a5afb296cab9d0ed8313a45ed8ba17967ecfca"},
 
    {file = "platformdirs-2.4.1.tar.gz", hash = "sha256:440633ddfebcc36264232365d7840a970e75e1018d15b4327d11f91909045fda"},
 
]
 
pygments = [
 
    {file = "Pygments-2.11.2-py3-none-any.whl", hash = "sha256:44238f1b60a76d78fc8ca0528ee429702aae011c265fe6a8dd8b63049ae41c65"},
 
    {file = "Pygments-2.11.2.tar.gz", hash = "sha256:4e426f72023d88d03b2fa258de560726ce890ff3b630f88c21cbb8b2503b8c6a"},
 
]
 
requests = [
 
    {file = "requests-2.27.1-py2.py3-none-any.whl", hash = "sha256:f22fa1e554c9ddfd16e6e41ac79759e17be9e492b3587efa038054674760e72d"},
 
    {file = "requests-2.27.1.tar.gz", hash = "sha256:68d7c56fd5a8999887728ef304a6d12edc7be74f1cfa47714fc8b414525c9a61"},
 
]
 
rich = [
 
    {file = "rich-11.2.0-py3-none-any.whl", hash = "sha256:d5f49ad91fb343efcae45a2b2df04a9755e863e50413623ab8c9e74f05aee52b"},
 
    {file = "rich-11.2.0.tar.gz", hash = "sha256:1a6266a5738115017bb64a66c59c717e7aa047b3ae49a011ede4abdeffc6536e"},
 
]
 
tomli = [
 
    {file = "tomli-2.0.0-py3-none-any.whl", hash = "sha256:b5bde28da1fed24b9bd1d4d2b8cba62300bfb4ec9a6187a957e8ddb9434c5224"},
 
    {file = "tomli-2.0.0.tar.gz", hash = "sha256:c292c34f58502a1eb2bbb9f5bbc9a5ebc37bee10ffb8c2d6bbdfa8eb13cc14e1"},
pyproject.toml
Show inline comments
 
@@ -8,6 +8,7 @@ authors = ["Dennis Fink <dennis.fink@c3l
 
python = "^3.10"
 
requests = "^2.27.1"
 
click = "^8.0.3"
 
rich = "^11.2.0"
 

	
 
[tool.poetry.dev-dependencies]
 
mypy = "^0.931"
stockcli/__init__.py
Show inline comments
 
new file 100644
stockcli/cli.py
Show inline comments
 
new file 100644
 
import json
 
import logging.config
 
from operator import itemgetter
 

	
 
import click
 
import requests
 
from rich.panel import Panel
 
from rich.table import Table
 

	
 
from .console import DEFAULT_PADDING, console, int_prompt, prompt
 
from .stock import (
 
    add_by_barcode,
 
    get_info_by_barcode,
 
    transfer_by_barcode,
 
    update_by_barcode,
 
)
 
from .utils import prepare_barcode
 

	
 
TASK_MAP = {
 
    "1": ("Transfer stock", transfer_by_barcode),
 
    "2": ("Add stock", add_by_barcode),
 
    "3": ("Update stock", update_by_barcode),
 
    "4": ("Get product info", get_info_by_barcode),
 
}
 

	
 

	
 
@click.command(context_settings={"help_option_names": ("-h", "--help", "-?")})
 
@click.option(
 
    "-c",
 
    "--config",
 
    "configfile",
 
    default="/etc/stockcli.json",
 
    type=click.File("r"),
 
    help="Config file to load",
 
)
 
@click.pass_context
 
def stockcli(ctx: click.Context, configfile: click.File) -> None:
 

	
 
    config = json.load(configfile)  # type: ignore
 
    logging.config.dictConfig(config["logging"])
 

	
 
    ctx.ensure_object(dict)
 
    ctx.obj["request_session"] = requests.Session()
 
    ctx.obj["request_session"].headers.update(
 
        {
 
            "accept": "application/json",
 
            "GROCY-API-KEY": config["grocy"]["apikey"],
 
        }
 
    )
 
    ctx.obj["base_url"] = f"{config['grocy']['url']}/api/"
 

	
 
    menu = Table.grid(padding=DEFAULT_PADDING)
 
    menu.add_column(justify="left", style="green", no_wrap=True)
 
    menu.add_column(justify="left", style="cyan", no_wrap=True)
 

	
 
    for task_id, task in sorted(TASK_MAP.items(), key=itemgetter(0)):
 
        menu.add_row(task_id, task[0])
 

	
 
    while True:
 
        click.clear()
 
        console.print(Panel(menu, title="[green bold]Menu[/green bold]"))
 
        choice = prompt.ask(
 
            "Enter a number to select a task", choices=list(TASK_MAP.keys())
 
        )
 
        selected_task = TASK_MAP[choice][1]
 

	
 
        rerun = True
 
        while rerun:
 
            barcode = prompt.ask("Please scan the barcode")
 
            barcode = prepare_barcode(barcode)
 

	
 
            try:
 
                selected_task(barcode)
 
            except (
 
                requests.Timeout,
 
                requests.ConnectionError,
 
                requests.HTTPError,
 
                requests.TooManyRedirects,
 
            ):
 
                rerun = False
 
                console.input("Continue")
 
            else:
 
                rerun = bool(
 
                    int_prompt.ask(
 
                        "Do the same with another product",
 
                        choices=["0", "1"],
 
                        default=0,
 
                    )
 
                )
 

	
 

	
 
if __name__ == "__main__":
 
    stockcli()
stockcli/console.py
Show inline comments
 
new file 100644
 
from rich.console import Console
 
from rich.prompt import IntPrompt, Prompt
 

	
 
DEFAULT_PADDING = (0, 1, 0, 0)
 
console = Console()
 
error_console = Console(stderr=True, style="bold red")
 
prompt = Prompt(console=console)
 
int_prompt = IntPrompt(console=console)
stockcli/stock.py
Show inline comments
 
new file 100644
 
from operator import itemgetter
 

	
 
from rich.panel import Panel
 
from rich.table import Table
 

	
 
from . import utils
 
from .console import DEFAULT_PADDING, console, int_prompt, prompt
 
from .style import GreenBoldText
 

	
 

	
 
def get_info_by_barcode(barcode: str) -> None:
 

	
 
    product = utils.get_request(f"stock/products/by-barcode/{barcode}")
 

	
 
    inner_product = product["product"]
 
    product_id = inner_product["id"]
 

	
 
    product_group = utils.get_request(
 
        f"objects/product_groups/{inner_product['product_group_id']}"
 
    )
 

	
 
    grid = Table.grid(padding=DEFAULT_PADDING)
 
    grid.add_column(justify="left", no_wrap=True)
 
    grid.add_column(justify="right", style="cyan", no_wrap=True)
 
    grid.add_column(justify="left", no_wrap=True)
 

	
 
    grid.add_row(GreenBoldText("Product ID:"), product_id)
 
    grid.add_row(GreenBoldText("Product Name:"), inner_product["name"])
 
    grid.add_row(GreenBoldText("Product Group:"), product_group["name"])
 

	
 
    for i, barcode_data in enumerate(product["product_barcodes"], start=1):
 
        grid.add_row(
 
            GreenBoldText(f"Barcode N°{i}:"),
 
            barcode_data["barcode"],
 
            barcode_data["note"],
 
        )
 

	
 
    purchase_quantity_unit = product["default_quantity_unit_purchase"]
 
    stock_quantity_unit = product["quantity_unit_stock"]
 

	
 
    purchase_to_stock_conversion = utils.get_request(
 
        f"objects/quantity_unit_conversions?query[]=from_qu_id={purchase_quantity_unit['id']}&query[]=to_qu_id={stock_quantity_unit['id']}"
 
    )
 

	
 
    if len(purchase_to_stock_conversion) != 0:
 
        conversion = f"{purchase_to_stock_conversion[0]['factor']} {stock_quantity_unit['name_plural']}"
 
    else:
 
        conversion = ""
 

	
 
    grid.add_row(
 
        GreenBoldText("Purchase Quantity Unit:"),
 
        purchase_quantity_unit["name"],
 
        conversion,
 
    )
 
    grid.add_row(
 
        GreenBoldText("Stock Quantity Unit:"),
 
        stock_quantity_unit["name"],
 
    )
 

	
 
    stock_amount = product["stock_amount"]
 
    grid.add_row(GreenBoldText("Overall Stock Amount:"), str(stock_amount))
 

	
 
    if stock_amount > 0:
 
        grid.add_row(GreenBoldText("Locations:"))
 
        locations = utils.get_request(f"stock/products/{product_id}/locations")
 

	
 
        for location in sorted(locations, key=itemgetter("location_name")):
 
            grid.add_row(
 
                f"[green bold]-[/green bold] [magenta]{location['location_name']}[/magenta]",
 
                location["amount"],
 
            )
 

	
 
    console.print(Panel(grid, title="[green bold]Product Info[/green bold]"))
 

	
 
    return
 

	
 

	
 
def transfer_by_barcode(barcode: str) -> None:
 

	
 
    product = utils.get_request(f"stock/products/by-barcode/{barcode}")
 

	
 
    inner_product = product["product"]
 
    product_id = inner_product["id"]
 

	
 
    locations = utils.get_request(f"stock/products/{product_id}/locations")
 
    all_locations = utils.get_request("objects/locations")
 

	
 
    grid = Table.grid(padding=DEFAULT_PADDING)
 
    grid.add_column(justify="right", style="green", no_wrap=True)
 
    grid.add_column(justify="left", style="cyan", no_wrap=True)
 
    grid.add_column(justify="right", style="magenta", no_wrap=True)
 

	
 
    grid.add_row("Product ID:", product_id, "")
 
    grid.add_row("Product Name:", inner_product["name"], "")
 
    grid.add_row("Locations:", "", "")
 

	
 
    choices = []
 

	
 
    for location in sorted(locations, key=itemgetter("location_id")):
 
        location_id = location["location_id"]
 
        choices.append(location_id)
 
        grid.add_row(
 
            f"[blue]{location_id}[/blue]",
 
            location["location_name"],
 
            location["amount"],
 
        )
 

	
 
    choices.append("0")
 
    console.print(
 
        Panel(
 
            grid,
 
            title="[green bold]Product Info[/green bold]",
 
            subtitle="[blue]Transfer stock[/blue]",
 
        )
 
    )
 

	
 
    from_id = int_prompt.ask(
 
        "From",
 
        choices=choices,
 
        default=int(choices[0]),
 
    )
 

	
 
    if not from_id:
 
        return
 

	
 
    grid = Table.grid(padding=DEFAULT_PADDING)
 
    grid.add_column(justify="right", style="green", no_wrap=True)
 
    grid.add_column(justify="left", style="cyan", no_wrap=True)
 

	
 
    choices = []
 
    for location in sorted(all_locations, key=itemgetter("id")):
 
        location_id = location["id"]
 
        if int(location_id) != from_id:
 
            choices.append(location_id)
 
            grid.add_row(f"[blue]{location_id}[/blue]", location["name"])
 

	
 
    choices.append("0")
 
    console.print(
 
        Panel(
 
            grid,
 
            title="[green bold]Locations[/green bold]",
 
            subtitle="[blue]Transfer Stock[/blue]",
 
        )
 
    )
 

	
 
    to_id = int_prompt.ask(
 
        "To",
 
        choices=choices,
 
        default=int(choices[0]),
 
    )
 

	
 
    if not to_id:
 
        return
 

	
 
    amount = int_prompt.ask("Amount")
 

	
 
    if not amount:
 
        return
 

	
 
    data = {"amount": amount, "location_id_from": from_id, "location_id_to": to_id}
 

	
 
    response = utils.post_request(f"stock/products/by-barcode/{barcode}/transfer", data)
 
    console.print("Successfully transfered!")
 
    return
 

	
 

	
 
def add_by_barcode(barcode: str) -> None:
 

	
 
    product = utils.get_request(f"stock/products/by-barcode/{barcode}")
 

	
 
    inner_product = product["product"]
 
    product_id = inner_product["id"]
 

	
 
    grid = Table.grid(padding=DEFAULT_PADDING)
 
    grid.add_column(justify="right", style="green", no_wrap=True)
 
    grid.add_column(justify="left", style="cyan", no_wrap=True)
 
    grid.add_row("Product ID:", product_id)
 
    grid.add_row("Product Name:", inner_product["name"])
 
    console.print(
 
        Panel(
 
            grid,
 
            title="[green bold]Product Info[/green bold]",
 
            subtitle="[blue]Add Stock[/blue]",
 
        )
 
    )
 

	
 
    amount = int_prompt.ask("Amount")
 
    data = {"amount": amount, "transaction_type": "purchase"}
 

	
 
    response = utils.post_request(f"stock/products/by-barcode/{barcode}/add", data)
 
    console.print("Successfully added!")
 
    return
 

	
 

	
 
def update_by_barcode(barcode: str) -> None:
 

	
 
    product = utils.get_request(f"stock/products/by-barcode/{barcode}")
 

	
 
    inner_product = product["product"]
 
    product_id = inner_product["id"]
 

	
 
    grid = Table.grid(padding=DEFAULT_PADDING)
 
    grid.add_column(justify="right", style="green", no_wrap=True)
 
    grid.add_column(justify="left", style="cyan", no_wrap=True)
 

	
 
    grid.add_row("Product ID:", product_id)
 
    grid.add_row("Product Name:", inner_product["name"])
 
    grid.add_row("Overall Stock Amount:", product["stock_amount"])
 

	
 
    console.print(
 
        Panel(
 
            grid,
 
            title="[green bold]Product Info[/green bold]",
 
            subtitle="[blue]Update stock[/blue]",
 
        )
 
    )
 
    amount = prompt.ask("Amount")
 

	
 
    data = {
 
        "new_amount": amount,
 
    }
 

	
 
    response = utils.post_request(f"stock/products/{product_id}/inventory", data)
 
    console.print("Successfully updated!")
 
    return
stockcli/style.py
Show inline comments
 
new file 100644
 
from functools import partial
 

	
 
from rich.style import Style
 
from rich.text import Text
 

	
 
GREEN_BOLD = Style(color="green", bold=True)
 

	
 
GreenBoldText = partial(Text, style=GREEN_BOLD)
stockcli/utils.py
Show inline comments
 
new file 100644
 
import logging
 
import string
 
from operator import attrgetter
 
from typing import Any, Optional
 

	
 
import click
 
import requests
 

	
 
from .console import error_console
 

	
 
UNALLOWED_CHARACTERS = str.maketrans(dict((c, None) for c in string.whitespace))
 

	
 

	
 
def make_request(method: str, url_path: str, data: Optional[Any] = None) -> Any:
 
    obj = click.get_current_context().obj
 
    session = obj["request_session"]
 
    base_url = obj["base_url"]
 
    requested_url = base_url + url_path
 

	
 
    method_function = attrgetter(method)
 

	
 
    try:
 
        if data is not None:
 
            response = method_function(session)(requested_url, json=data)
 
        else:
 
            response = method_function(session)(requested_url)
 
        response.raise_for_status()
 
    except requests.Timeout:
 
        logging.error(f"The connection to {requested_url} timed out!")
 
        error_console.print("Connection timed out!")
 
        raise
 
    except requests.ConnectionError:
 
        logging.error(f"Couldn't establish a connection to {requested_url}!")
 
        error_console.print("Couldn't establish a connection!")
 
        raise
 
    except requests.HTTPError:
 
        logging.error(f"{requested_url} sent back an HTTPError")
 
        error_console.print("Got the following error:")
 
        error_message = response.json()["error_message"]
 
        logging.error(error_message)
 
        error_console.print(error_message)
 
        raise
 
    except requests.TooManyRedirects:
 
        logging.error(f"{requested_url} had too many redirects!")
 
        error_console.print("Too many redirects!")
 
        raise
 
    else:
 
        return response.json()
 

	
 

	
 
def get_request(url_path: str) -> Any:
 
    return make_request("get", url_path)
 

	
 

	
 
def post_request(url_path: str, data: dict[str, Any]) -> Any:
 
    return make_request("post", url_path, data)
 

	
 

	
 
def put_request(url_path: str, data: dict[str, Any]) -> Any:
 
    return make_request("put", url_path, data)
 

	
 

	
 
def prepare_barcode(barcode: str) -> str:
 
    return barcode.translate(UNALLOWED_CHARACTERS)
0 comments (0 inline, 0 general)