88 lines
3 KiB
Python
88 lines
3 KiB
Python
#!/usr/bin/env python3
|
|
"""Manage prodev expense entries in wip.agent_tasks on DocDB.
|
|
|
|
Usage:
|
|
prodev_tasks.py create "<note>" [--price PRICE] [--vendor VENDOR] [--product-type TYPE]
|
|
prodev_tasks.py update <document_id> "<note>" [--price PRICE] [--vendor VENDOR] [--product-type TYPE]
|
|
prodev_tasks.py list
|
|
"""
|
|
|
|
import sys
|
|
import argparse
|
|
from pymongo import MongoClient
|
|
|
|
URI = "mongodb://root:3wwfoUjyk2E2zWELXFlLuHqfw1ALlOp4pb2H5Vq3TImbMIHL2h1u8Jej2mjzCPl@docdb.connorrhodes.com:35563?tls=true&tlsAllowInvalidCertificates=true"
|
|
DB = "wip"
|
|
COLL = "agent_tasks"
|
|
|
|
def get_collection():
|
|
client = MongoClient(URI, serverSelectionTimeoutMS=10000)
|
|
return client[DB][COLL], client
|
|
|
|
def create(note, price=None, vendor=None, product_type=None):
|
|
coll, client = get_collection()
|
|
doc = {"note": note, "type": "prodev expense", "status": "todo"}
|
|
if price is not None:
|
|
doc["price"] = float(price)
|
|
if vendor is not None:
|
|
doc["vendor"] = vendor
|
|
if product_type is not None:
|
|
doc["product_type"] = product_type
|
|
result = coll.insert_one(doc)
|
|
print(result.inserted_id)
|
|
client.close()
|
|
|
|
def update(doc_id, note, price=None, vendor=None, product_type=None):
|
|
coll, client = get_collection()
|
|
sets = {"note": note}
|
|
if price is not None:
|
|
sets["price"] = float(price)
|
|
if vendor is not None:
|
|
sets["vendor"] = vendor
|
|
if product_type is not None:
|
|
sets["product_type"] = product_type
|
|
result = coll.update_one({"_id": doc_id}, {"$set": sets})
|
|
print(f"matched={result.matched_count} modified={result.modified_count}")
|
|
client.close()
|
|
|
|
def list_all():
|
|
from bson import ObjectId
|
|
coll, client = get_collection()
|
|
for doc in coll.find({"type": "prodev expense"}):
|
|
price = doc.get("price", "?")
|
|
vendor = doc.get("vendor", "?")
|
|
ptype = doc.get("product_type", "?")
|
|
print(f"{doc['_id']} | ${price} | {vendor} | {ptype} | {doc['status']} | {doc['note'][:60]}...")
|
|
client.close()
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Manage prodev expense entries")
|
|
sub = parser.add_subparsers(dest="cmd")
|
|
|
|
p_create = sub.add_parser("create")
|
|
p_create.add_argument("note")
|
|
p_create.add_argument("--price", default=None)
|
|
p_create.add_argument("--vendor", default=None)
|
|
p_create.add_argument("--product-type", default=None)
|
|
|
|
p_update = sub.add_parser("update")
|
|
p_update.add_argument("doc_id")
|
|
p_update.add_argument("note")
|
|
p_update.add_argument("--price", default=None)
|
|
p_update.add_argument("--vendor", default=None)
|
|
p_update.add_argument("--product-type", default=None)
|
|
|
|
sub.add_parser("list")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.cmd == "create":
|
|
create(args.note, args.price, args.vendor, args.product_type)
|
|
elif args.cmd == "update":
|
|
from bson import ObjectId
|
|
update(ObjectId(args.doc_id), args.note, args.price, args.vendor, args.product_type)
|
|
elif args.cmd == "list":
|
|
list_all()
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(1)
|