mirror of
https://github.com/dtomlinson91/tembo.git
synced 2025-12-22 04:05:45 +00:00
adding latest
This commit is contained in:
@@ -6,7 +6,7 @@ from panaetius.exceptions import LoggingDirectoryDoesNotExistException
|
||||
__version__ = "0.1.0"
|
||||
|
||||
if (config_path := os.environ.get("TEMBO_CONFIG")) is not None:
|
||||
CONFIG = panaetius.Config("tembo", config_path)
|
||||
CONFIG = panaetius.Config("tembo", config_path, skip_header_init=True)
|
||||
else:
|
||||
CONFIG = panaetius.Config("tembo", "~/tembo/.config", skip_header_init=True)
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import click
|
||||
|
||||
import tembo.cli
|
||||
from tembo.journal import pages
|
||||
from tembo.utils import Success
|
||||
from tembo import exceptions
|
||||
|
||||
|
||||
@@ -61,57 +62,71 @@ def new(scope: str, inputs: Collection[str], dry_run: bool, example: bool):
|
||||
"""
|
||||
|
||||
# check that the name exists in the config.yml
|
||||
_cli_verify_name_exists(scope)
|
||||
try:
|
||||
_new_verify_name_exists(scope)
|
||||
except (
|
||||
exceptions.ScopeNotFound,
|
||||
exceptions.EmptyConfigYML,
|
||||
exceptions.MissingConfigYML,
|
||||
) as tembo_exception:
|
||||
cli_message(tembo_exception.args[0])
|
||||
raise SystemExit(0) from tembo_exception
|
||||
|
||||
# get the scope configuration from the config.yml
|
||||
config_scope = _cli_get_config_scope(scope)
|
||||
try:
|
||||
config_scope = _new_get_config_scope(scope)
|
||||
except exceptions.MandatoryKeyNotFound as mandatory_key_not_found:
|
||||
cli_message(mandatory_key_not_found.args[0])
|
||||
raise SystemExit(1) from mandatory_key_not_found
|
||||
|
||||
# if --example flag, return the example to the user
|
||||
_cli_show_example(example, config_scope)
|
||||
_new_show_example(example, config_scope)
|
||||
|
||||
# if the name is in the config.yml, create the scoped page
|
||||
scoped_page = _cli_create_scoped_page(config_scope, inputs)
|
||||
scoped_page = _new_create_scoped_page(config_scope, inputs)
|
||||
|
||||
if dry_run:
|
||||
cli_message(f"{scoped_page.path} will be created")
|
||||
raise SystemExit(0)
|
||||
|
||||
try:
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
if isinstance(result, Success):
|
||||
cli_message(f"Saved {result.message} to disk")
|
||||
raise SystemExit(0)
|
||||
except exceptions.ScopedPageAlreadyExists as scoped_page_already_exists:
|
||||
cli_message(f"File {scoped_page_already_exists}")
|
||||
raise SystemExit(0) from scoped_page_already_exists
|
||||
|
||||
|
||||
def _cli_verify_name_exists(scope: str) -> None:
|
||||
_name_found = scope in [user_scope["name"] for user_scope in tembo.cli.CONFIG.scopes]
|
||||
def _new_verify_name_exists(scope: str) -> None:
|
||||
_name_found = scope in [
|
||||
user_scope["name"] for user_scope in tembo.cli.CONFIG.scopes
|
||||
]
|
||||
if _name_found:
|
||||
return
|
||||
if len(tembo.cli.CONFIG.scopes) > 0:
|
||||
# if the name is missing in the config.yml, raise error
|
||||
cli_message(f"Command {scope} not found in config.yml.")
|
||||
raise SystemExit(0)
|
||||
raise exceptions.ScopeNotFound(f"Command {scope} not found in config.yml")
|
||||
# raise error if no config.yml found
|
||||
if pathlib.Path(tembo.cli.CONFIG.config_path).exists():
|
||||
tembo.cli.logger.critical(
|
||||
"Config.yml found in %s is empty - exiting", tembo.cli.CONFIG.config_path
|
||||
raise exceptions.EmptyConfigYML(
|
||||
f"Config.yml found in {tembo.cli.CONFIG.config_path} is empty"
|
||||
)
|
||||
else:
|
||||
tembo.cli.logger.critical(
|
||||
"No config.yml found in %s - exiting", tembo.cli.CONFIG.config_path
|
||||
)
|
||||
raise SystemExit(1)
|
||||
raise exceptions.MissingConfigYML(
|
||||
f"No config.yml found in {tembo.cli.CONFIG.config_path}"
|
||||
)
|
||||
|
||||
|
||||
def _cli_get_config_scope(scope: str) -> dict:
|
||||
def _new_get_config_scope(scope: str) -> dict:
|
||||
config_scope = {}
|
||||
optional_keys = ["example", "template_filename"]
|
||||
for option in [
|
||||
"name",
|
||||
"example",
|
||||
"path",
|
||||
"filename",
|
||||
"extension",
|
||||
"example",
|
||||
"template_filename",
|
||||
]:
|
||||
try:
|
||||
@@ -123,17 +138,19 @@ def _cli_get_config_scope(scope: str) -> dict:
|
||||
}
|
||||
)
|
||||
except KeyError as key_error:
|
||||
if key_error.args[0] in ["example", "template_filename"]:
|
||||
if key_error.args[0] in optional_keys:
|
||||
config_scope.update({key_error.args[0]: None})
|
||||
continue
|
||||
tembo.cli.logger.critical(
|
||||
"Key %s not found in config. yml - exiting", key_error
|
||||
)
|
||||
raise SystemExit(1) from key_error
|
||||
raise exceptions.MandatoryKeyNotFound(
|
||||
f"Key {key_error} not found in config.yml"
|
||||
)
|
||||
return config_scope
|
||||
|
||||
|
||||
def _cli_show_example(example: bool, config_scope: dict) -> None:
|
||||
def _new_show_example(example: bool, config_scope: dict) -> None:
|
||||
if example:
|
||||
tembo.cli.logger.info(
|
||||
"Example for 'tembo new %s': %s",
|
||||
@@ -145,9 +162,10 @@ def _cli_show_example(example: bool, config_scope: dict) -> None:
|
||||
raise SystemExit(0)
|
||||
|
||||
|
||||
def _cli_create_scoped_page(config_scope: dict, inputs: Collection[str]) -> pages.Page:
|
||||
def _new_create_scoped_page(config_scope: dict, inputs: Collection[str]) -> pages.Page:
|
||||
page_creator_options = pages.PageCreatorOptions(
|
||||
base_path=tembo.cli.CONFIG.base_path,
|
||||
template_path=tembo.cli.CONFIG.template_path,
|
||||
page_path=config_scope["path"],
|
||||
filename=config_scope["filename"],
|
||||
extension=config_scope["extension"],
|
||||
@@ -155,7 +173,6 @@ def _cli_create_scoped_page(config_scope: dict, inputs: Collection[str]) -> page
|
||||
example=config_scope["example"],
|
||||
user_input=inputs,
|
||||
template_filename=config_scope["template_filename"],
|
||||
template_path=tembo.cli.CONFIG.template_path,
|
||||
)
|
||||
try:
|
||||
return pages.ScopedPageCreator(page_creator_options).create_page()
|
||||
@@ -191,9 +208,7 @@ run.add_command(list_all)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# new(["meeting", "robs presentation", "meeting on gcp"])
|
||||
new(["meeting", "a", "b", "c", "d"]) # noqa
|
||||
# new(["meeting", "robs presentation"])
|
||||
|
||||
# pyinstaller
|
||||
# if getattr(sys, "frozen", False):
|
||||
|
||||
@@ -18,3 +18,19 @@ class TemplateFileNotFoundError(Exception):
|
||||
|
||||
class ScopedPageAlreadyExists(Exception):
|
||||
"""Raised if the scoped page file already exists."""
|
||||
|
||||
|
||||
class MissingConfigYML(Exception):
|
||||
"""Raised if the config.yml file is missing."""
|
||||
|
||||
|
||||
class EmptyConfigYML(Exception):
|
||||
"""Raised if the config.yml file is empty."""
|
||||
|
||||
|
||||
class ScopeNotFound(Exception):
|
||||
"""Raised if the scope does not exist in the config.yml."""
|
||||
|
||||
|
||||
class MandatoryKeyNotFound(Exception):
|
||||
"""Raised if a mandatory key is not found in the config.yml."""
|
||||
|
||||
@@ -12,6 +12,7 @@ from jinja2.exceptions import TemplateNotFound
|
||||
|
||||
import tembo
|
||||
from tembo import exceptions
|
||||
import tembo.utils
|
||||
|
||||
|
||||
# TODO: flesh this out with details for the optional args
|
||||
@@ -222,7 +223,7 @@ class Page(metaclass=ABCMeta):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def save_to_disk(self) -> None:
|
||||
def save_to_disk(self) -> tembo.utils.Success:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
@@ -245,13 +246,13 @@ class ScopedPage(Page):
|
||||
self.page_content = page_content
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"ScopedPage({self.path})"
|
||||
return f"ScopedPage(\"{self.path}\")"
|
||||
|
||||
@property
|
||||
def path(self) -> pathlib.Path:
|
||||
return self._path
|
||||
|
||||
def save_to_disk(self) -> None:
|
||||
def save_to_disk(self) -> tembo.utils.Success:
|
||||
"""Save the scoped page to disk and write the `page_content`.
|
||||
|
||||
If the page already exists a message will be logged to stdout and no file
|
||||
@@ -274,8 +275,6 @@ class ScopedPage(Page):
|
||||
scoped_page_file.parents[0].mkdir(parents=True, exist_ok=True)
|
||||
if scoped_page_file.exists():
|
||||
raise exceptions.ScopedPageAlreadyExists(f"{self.path} already exists")
|
||||
|
||||
with scoped_page_file.open("w", encoding="utf-8") as scoped_page:
|
||||
scoped_page.write(self.page_content)
|
||||
# TODO: pass this back somehow
|
||||
tembo.cli.cli.logger.info("Saved %s to disk", self.path)
|
||||
return tembo.utils.Success(str(self.path))
|
||||
|
||||
12
tembo/utils/__init__.py
Normal file
12
tembo/utils/__init__.py
Normal file
@@ -0,0 +1,12 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Success:
|
||||
"""Success message.
|
||||
|
||||
Attributes:
|
||||
message (str): A success message.
|
||||
"""
|
||||
|
||||
message: str
|
||||
0
tests/test_cli/__init__.py
Normal file
0
tests/test_cli/__init__.py
Normal file
7
tests/test_cli/data/config/success/config.yml
Normal file
7
tests/test_cli/data/config/success/config.yml
Normal file
@@ -0,0 +1,7 @@
|
||||
tembo:
|
||||
scopes:
|
||||
- name: some_scope
|
||||
example: tembo new some_scope
|
||||
path: "some_scope"
|
||||
filename: "{name}"
|
||||
extension: md
|
||||
50
tests/test_cli/test_cli.py
Normal file
50
tests/test_cli/test_cli.py
Normal file
@@ -0,0 +1,50 @@
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
import tembo.exceptions
|
||||
|
||||
|
||||
def test_cli_page_is_saved_success():
|
||||
pass
|
||||
|
||||
|
||||
def test_new_verify_name_exists_success(shared_datadir):
|
||||
# arrange
|
||||
os.environ["TEMBO_CONFIG"] = str(shared_datadir / "config" / "success")
|
||||
import tembo.cli
|
||||
from tembo.cli.cli import _new_verify_name_exists
|
||||
|
||||
c = tembo.cli.CONFIG
|
||||
|
||||
# act
|
||||
verified_name = _new_verify_name_exists("some_scope")
|
||||
|
||||
# assert
|
||||
assert verified_name is None
|
||||
|
||||
|
||||
def test_new_verify_name_exists_scope_not_found(shared_datadir):
|
||||
# arrange
|
||||
os.environ["TEMBO_CONFIG"] = str(shared_datadir / "config" / "success")
|
||||
import tembo.cli
|
||||
from tembo.cli.cli import _new_verify_name_exists
|
||||
|
||||
c = tembo.cli.CONFIG
|
||||
|
||||
# act
|
||||
with pytest.raises(tembo.exceptions.ScopeNotFound) as scope_not_found:
|
||||
_new_verify_name_exists("some_missing_scope")
|
||||
|
||||
# assert
|
||||
assert str(scope_not_found.value) == "Command some_missing_scope not found in config.yml"
|
||||
|
||||
|
||||
def test_new_get_config_scope(shared_datadir):
|
||||
# arrange
|
||||
os.environ["TEMBO_CONFIG"] = str(shared_datadir / "config" / "success")
|
||||
import tembo.cli
|
||||
|
||||
# act
|
||||
|
||||
# assert
|
||||
@@ -5,6 +5,7 @@ import pytest
|
||||
|
||||
from tembo import PageCreatorOptions, ScopedPageCreator
|
||||
from tembo import exceptions
|
||||
from tembo.utils import Success
|
||||
|
||||
|
||||
DATE_TODAY = date.today().strftime("%d-%m-%Y")
|
||||
@@ -90,7 +91,7 @@ def test_create_page_already_exists(datadir):
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
with pytest.raises(exceptions.ScopedPageAlreadyExists) as page_already_exists:
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
@@ -99,7 +100,7 @@ def test_create_page_already_exists(datadir):
|
||||
assert scoped_page_contents.readlines() == ["this file already exists\n"]
|
||||
|
||||
|
||||
def test_create_page_without_template(tmpdir, caplog):
|
||||
def test_create_page_without_template(tmpdir):
|
||||
# arrange
|
||||
options = PageCreatorOptions(
|
||||
base_path=str(tmpdir),
|
||||
@@ -112,18 +113,18 @@ def test_create_page_without_template(tmpdir, caplog):
|
||||
template_filename=None,
|
||||
template_path=None,
|
||||
)
|
||||
# TODO: copy this pattern creation into the other tests
|
||||
scoped_page_file = (
|
||||
pathlib.Path(options.base_path) / options.page_path / options.filename
|
||||
).with_suffix(f".{options.extension}")
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
with scoped_page_file.open("r", encoding="utf-8") as scoped_page_contents:
|
||||
assert scoped_page_contents.readlines() == []
|
||||
|
||||
@@ -147,11 +148,12 @@ def test_create_page_with_template(datadir, caplog):
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
with scoped_page_file.open("r", encoding="utf-8") as scoped_page_contents:
|
||||
assert scoped_page_contents.readlines() == [
|
||||
"scoped page file\n",
|
||||
@@ -193,11 +195,12 @@ def test_create_tokened_page_tokens_in_template(
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
|
||||
with scoped_page_file.open("r", encoding="utf-8") as scoped_page_contents:
|
||||
assert scoped_page_contents.readline() == page_contents
|
||||
@@ -236,11 +239,12 @@ def test_create_tokened_page_tokens_in_filename(
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
|
||||
|
||||
def test_create_tokened_page_input_tokens_preserve_order(datadir, caplog):
|
||||
@@ -263,11 +267,12 @@ def test_create_tokened_page_input_tokens_preserve_order(datadir, caplog):
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
with scoped_page_file.open(mode="r", encoding="utf-8") as scoped_page_contents:
|
||||
assert scoped_page_contents.readline() == "third_input second_input"
|
||||
|
||||
@@ -324,11 +329,12 @@ def test_create_page_spaces_in_path(tmpdir, caplog):
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
|
||||
|
||||
def test_create_page_dot_in_extension(tmpdir, caplog):
|
||||
@@ -350,11 +356,12 @@ def test_create_page_dot_in_extension(tmpdir, caplog):
|
||||
|
||||
# act
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
scoped_page.save_to_disk()
|
||||
result = scoped_page.save_to_disk()
|
||||
|
||||
# assert
|
||||
assert scoped_page_file.exists()
|
||||
assert caplog.records[0].message == f"Saved {scoped_page_file} to disk"
|
||||
assert isinstance(result, Success)
|
||||
assert result.message == str(scoped_page_file)
|
||||
|
||||
|
||||
def test_create_page_str_representation(tmpdir):
|
||||
@@ -378,4 +385,4 @@ def test_create_page_str_representation(tmpdir):
|
||||
scoped_page = ScopedPageCreator(options).create_page()
|
||||
|
||||
# assert
|
||||
assert str(scoped_page) == f"ScopedPage({scoped_page_file})"
|
||||
assert str(scoped_page) == f"ScopedPage(\"{scoped_page_file}\")"
|
||||
|
||||
Reference in New Issue
Block a user