# Global variables
PROJECT_ROOT = os.path.abspath(os.path.dirname(__file__))
NUTRA_HOME = os.getenv("NUTRA_HOME", os.path.join(os.path.expanduser("~"), ".nutra"))
-USDA_DB_NAME = "usda.sqlite"
+USDA_DB_NAME = "usda.sqlite3"
# NOTE: NT_DB_NAME = "nt.sqlite3" is defined in ntclient.ntsqlite.sql
NTSQLITE_BUILDPATH = os.path.join(PROJECT_ROOT, "ntsqlite", "sql", NT_DB_NAME)
def bugs_list(args: argparse.Namespace) -> tuple:
"""List bug reports that have been saved"""
- _bugs_list = ntclient.services.bugs.list_bugs()
- n_bugs_total = len(_bugs_list)
- n_bugs_unsubmitted = len([x for x in _bugs_list if not bool(x[-1])])
+ rows, headers = ntclient.services.bugs.list_bugs()
+ n_bugs_total = len(rows)
+ n_bugs_unsubmitted = len([x for x in rows if not bool(x[-1])])
print(f"You have: {n_bugs_total} total bugs amassed in your journey.")
print(f"Of these, {n_bugs_unsubmitted} require submission/reporting.")
print()
- for bug in _bugs_list:
+ print(rows)
+ # print([[entry for entry in row] for row in rows])
+ # exit(0)
+ table = tabulate(
+ [[entry for entry in row if "\n" not in str(entry)] for row in rows],
+ headers=headers,
+ tablefmt="presto",
+ )
+ print(table)
+ exit(0)
+
+ for bug in rows:
if not args.show:
continue
# Skip submitted bugs by default
if bool(bug[-1]) and not args.debug:
continue
# Print all bug properties (except noisy stacktrace)
- print(", ".join(str(x) for x in bug if "\n" not in str(x)))
+ bug_line = str()
+ for _col_name, _value in dict(bug).items():
+ print(_col_name)
+ print(_value)
+ if "\n" in str(_value):
+ continue
+ bug_line += str(_value) + ", "
+ # print(", ".join(str(x) for x in bug if "\n" not in str(x)))
+ print()
if n_bugs_unsubmitted > 0:
print("NOTE: You have bugs awaiting submission. Please run the report command")
- return 0, _bugs_list
+ return 0, rows
# pylint: disable=unused-argument
import sqlite3
from collections.abc import Sequence
+from typing import Optional
from ntclient.utils import CLI_CONFIG
# ------------------------------------------------
-def sql_entries(sql_result: sqlite3.Cursor) -> list:
- """Formats and returns a `sql_result()` for console digestion and output"""
- # TODO: return object: metadata, command, status, errors, etc?
+def sql_entries(sql_result: sqlite3.Cursor) -> tuple[list, list, int, Optional[int]]:
+ """
+ Formats and returns a `sql_result()` for console digestion and output
+ FIXME: the IDs are not necessarily integers, but are unique.
- rows = sql_result.fetchall()
- return rows
+ TODO: return object: metadata, command, status, errors, etc?
+ """
+
+ return (
+ # rows
+ sql_result.fetchall(),
+ # headers
+ [x[0] for x in sql_result.description],
+ # row_count
+ sql_result.rowcount,
+ # last_row_id
+ sql_result.lastrowid,
+ )
def sql_entries_headers(sql_result: sqlite3.Cursor) -> tuple:
query: str,
db_name: str,
values: Sequence = (),
-) -> list:
+) -> tuple[list, list, int, Optional[int]]:
"""@param values: tuple | list"""
cur = _prep_query(con, query, db_name, values)
close_con_and_cur(con, cur)
return result
-
-
-def _sql_headers(
- con: sqlite3.Connection,
- query: str,
- db_name: str,
- values: Sequence = (),
-) -> tuple:
- """@param values: tuple | list"""
-
- cur = _prep_query(con, query, db_name, values)
-
- result = sql_entries_headers(cur)
-
- close_con_and_cur(con, cur)
- return result
import os
import sqlite3
from collections.abc import Sequence
+from typing import Optional
from ntclient import (
NT_DB_NAME,
NUTRA_HOME,
__db_target_nt__,
)
-from ntclient.persistence.sql import _sql, _sql_headers, version
+from ntclient.persistence.sql import _sql, version
from ntclient.utils.exceptions import SqlConnectError, SqlInvalidVersionError
raise SqlConnectError("ERROR: nt database doesn't exist, please run `nutra init`")
-def sql(query: str, values: Sequence = ()) -> list:
+def sql(query: str, values: Sequence = ()) -> tuple[list, list, int, Optional[int]]:
"""Executes a SQL command to nt.sqlite3"""
con = nt_sqlite_connect()
return _sql(con, query, db_name="nt", values=values)
-
-
-def sql_headers(query: str, values: Sequence = ()) -> tuple:
- """Executes a SQL command to nt.sqlite3"""
-
- con = nt_sqlite_connect()
- return _sql_headers(con, query, db_name="nt", values=values)
def sql_nt_next_index(table: str) -> int:
"""Used for previewing inserts"""
+ # TODO: parameterized queries
# noinspection SqlResolve
query = "SELECT MAX(id) as max_id FROM %s;" % table # nosec: B608
- return int(sql(query)[0]["max_id"])
+ rows, _, _, _ = sql(query)
+ return int(rows[0]["max_id"])
import tarfile
import urllib.request
from collections.abc import Sequence
+from typing import Optional
from ntclient import NUTRA_HOME, USDA_DB_NAME, __db_target_usda__
-from ntclient.persistence.sql import _sql, _sql_headers, version
+from ntclient.persistence.sql import _sql, version
from ntclient.utils.exceptions import SqlConnectError, SqlInvalidVersionError
return version(con)
-def sql(query: str, values: Sequence = (), version_check: bool = True) -> list:
+def sql(
+ query: str, values: Sequence = (), version_check: bool = True
+) -> tuple[list, list, int, Optional[int]]:
"""
Executes a SQL command to usda.sqlite3
# TODO: support argument: _sql(..., params=params, ...)
return _sql(con, query, db_name="usda", values=values)
-
-
-def sql_headers(query: str, values: Sequence = (), version_check: bool = True) -> tuple:
- """
- Executes a SQL command to usda.sqlite3 [WITH HEADERS]
-
- @param query: Input SQL query
- @param values: Union[tuple, list] Leave as empty tuple for no values,
- e.g. bare query. Populate a tuple for a single insert. And use a list for
- cur.executemany()
- @param version_check: Ignore mismatch version, useful for "meta" commands
- @return: List of selected SQL items
- """
-
- con = usda_sqlite_connect(version_check=version_check)
-
- # TODO: support argument: _sql(..., params=params, ...)
- return _sql_headers(con, query, db_name="usda", values=values)
"""usda.sqlite functions module"""
from ntclient import NUTR_ID_KCAL
-from ntclient.persistence.sql.usda import sql, sql_headers
+from ntclient.persistence.sql.usda import sql
################################################################################
"""Shows food groups"""
query = "SELECT * FROM fdgrp;"
- result = sql(query)
- return {x[0]: x for x in result}
+ rows, _, _, _ = sql(query)
+ return {x[0]: x for x in rows}
def sql_food_details(_food_ids: set = None) -> list: # type: ignore
food_ids = ",".join(str(x) for x in set(_food_ids))
query = query % food_ids
- return sql(query)
+ rows, _, _, _ = sql(query)
+ return rows
def sql_nutrients_overview() -> dict:
"""Shows nutrients overview"""
query = "SELECT * FROM nutrients_overview;"
- result = sql(query)
- return {x[0]: x for x in result}
+ rows, _, _, _ = sql(query)
+ return {x[0]: x for x in rows}
def sql_nutrients_details() -> tuple:
"""Shows nutrients 'details'"""
query = "SELECT * FROM nutrients_overview;"
- return sql_headers(query)
+ rows, headers, _, _ = sql(query)
+ return rows, headers
def sql_servings(_food_ids: set) -> list:
"""
# FIXME: support this kind of thing by library code & parameterized queries
food_ids = ",".join(str(x) for x in set(_food_ids))
- return sql(query % food_ids)
+ rows, _, _, _ = sql(query % food_ids)
+ return rows
def sql_analyze_foods(food_ids: set) -> list:
"""
# TODO: parameterized queries
food_ids_concat = ",".join(str(x) for x in set(food_ids))
- return sql(query % food_ids_concat)
+ rows, _, _, _ = sql(query % food_ids_concat)
+ return rows
################################################################################
ORDER BY
food_id;
"""
-
- return sql(query % (NUTR_ID_KCAL, nutrient_id))
+ # TODO: parameterized queries
+ rows, _, _, _ = sql(query % (NUTR_ID_KCAL, nutrient_id))
+ return rows
def sql_sort_foods(nutr_id: int) -> list:
ORDER BY
nut_data.nutr_val DESC;
"""
-
- return sql(query % nutr_id)
+ # TODO: parameterized queries
+ rows, _, _, _ = sql(query % nutr_id)
+ return rows
def sql_sort_foods_by_kcal(nutr_id: int) -> list:
ORDER BY
(nut_data.nutr_val / kcal.nutr_val) DESC;
"""
-
- return sql(query % nutr_id)
+ # TODO: parameterized queries
+ rows, _, _, _ = sql(query % nutr_id)
+ return rows
"version_usda_db_target": __db_target_usda__,
}
),
- # user_details
- "NOT_IMPLEMENTED",
+ # user_details (TODO: add user details)
+ None,
),
)
except sqlite3.IntegrityError as exc:
raise
-def list_bugs() -> list:
- """List all bugs."""
- sql_bugs = sql_nt("SELECT * FROM bug")
- return sql_bugs
+def list_bugs() -> tuple[list, list]:
+ """List all bugs, with headers."""
+ rows, headers, _, _ = sql_nt("SELECT * FROM bug")
+ return rows, headers
def submit_bugs() -> int:
"""Submit bug reports to developer, return n_submitted."""
# Gather bugs for submission
- sql_bugs = sql_nt("SELECT * FROM bug WHERE submitted = 0")
+ rows, _, _, _ = sql_nt("SELECT * FROM bug WHERE submitted = 0")
api_client = ntclient.services.api.ApiClient()
n_submitted = 0
- print(f"submitting {len(sql_bugs)} bug reports...")
- print("_" * len(sql_bugs))
+ print(f"submitting {len(rows)} bug reports...")
+ print("_" * len(rows))
- for bug in sql_bugs:
+ for bug in rows:
_res = api_client.post_bug(bug)
if CLI_CONFIG.debug:
print(_res.json())
def list_nutrients() -> tuple:
"""Lists out nutrients with basic details"""
- headers, nutrients = sql_nutrients_details()
+ nutrients, headers = sql_nutrients_details()
# TODO: include in SQL table cache?
headers.append("avg_rda")
nutrients = [list(x) for x in nutrients]
projects / nutratech / cli.git / commitdiff