Add async pieces.

This commit is contained in:
hackish 2023-07-31 15:57:05 -07:00
parent 79e06ea46f
commit 32f9588e26
11 changed files with 241 additions and 37 deletions

View file

@ -0,0 +1,46 @@
import argparse
import asyncio
import sys
import uvloop
__version__ = "0.0.1"
async def main(args: argparse.Namespace) -> None:
"""
Main function of the program (stub).
This function serves as ${REPO_NAME_SNAKE}'s critical path for primary logic.
It receives an `argparse.Namespace` object as input from invoke_maiun, which includes
the command-line arguments that were defined and parsed in the `run` function.
Args:
args (argparse.Namespace): The command-line arguments object.
Returns:
None
"""
pass
def invoke_main(args: argparse.Namespace) -> None:
"""
Function to invoke the main function using asyncio and uvloop (stub).
This function serves as a bridge to run the async main function in
a uvloop event loop. It handles the differences in asyncio API across
different Python versions.
Args:
args (argparse.Namespace): The command-line arguments object.
Returns:
None
"""
if sys.version_info >= (3, 11):
with asyncio.Runner(loop_factory=uvloop.new_event_loop) as runner:
runner.run(main(args))
else:
uvloop.install()
asyncio.run(main(args))

44
${REPO_NAME_SNAKE}/cli.py Normal file
View file

@ -0,0 +1,44 @@
from argparse import ArgumentParser
from pathlib import Path
from . import __version__
from .logger import logger
from .validation import validate_file_folder
def run() -> None:
"""
Command-line interface for ${REPO_NAME_SNAKE}.
This function defines and handles command-line arguments for the program.
The available options are:
--version: Show the version of the program and exit.
--verbose: Enable verbose logging. This will produce detailed output messages.
--output/-o: Specify the output file. This must be a valid file path.
paths: Specify any number of existing files or directories to be processed.
These paths must point to existing files or directories.
Upon parsing the command-line arguments, the function configures the logger
and invokes the `main` function with the parsed arguments.
Returns:
None
"""
# your code here
parser = ArgumentParser(description="${REPO_DESCRIPTION}")
parser.add_argument("--version", action="version", version=__version__)
parser.add_argument("--verbose", action="store_true", help="Enable verbose logging")
parser.add_argument("--output", "-o", type=Path, help="Specify output file")
parser.add_argument(
"paths",
nargs="+",
type=validate_file_folder,
help="Specify any number of existing files or directories to be processed.",
)
args = parser.parse_args()
if args.verbose:
logger.debug("Verbose logging enabled")
run(args)

View file

@ -0,0 +1,4 @@
import logging
logging.basicConfig(format="%(asctime)s - %(message)s", datefmt="%d-%b-%y %H:%M:%S")
logger = logging.getLogger(__name__)

View file

@ -0,0 +1,12 @@
from pathlib import Path
from aiopath import AsyncPath
def validate_file_folder(value: str) -> Path:
file_folder_path = Path(value)
if not file_folder_path.exists():
raise FileNotFoundError(f"No such file or folder: {value}")
if not file_folder_path.is_file() and not file_folder_path.is_dir():
raise TypeError(f"Not a file or directory: {value}")
return AsyncPath(value)