{/* The gradient overlay to partially cover the title */}
Fable
);
}
```
### backend/env/lib/python3.12/site-packages/fastapi/cli.py
```python
try:
from fastapi_cli.cli import main as cli_main
except ImportError: # pragma: no cover
cli_main = None # type: ignore
def main() -> None:
if not cli_main: # type: ignore[truthy-function]
message = 'To use the fastapi command, please install "fastapi[standard]":\n\n\tpip install "fastapi[standard]"\n'
print(message)
raise RuntimeError(message) # noqa: B904
cli_main()
```
### backend/env/lib/python3.12/site-packages/dotenv/cli.py
```python
import json
import os
import shlex
import sys
from contextlib import contextmanager
from subprocess import Popen
from typing import Any, Dict, IO, Iterator, List
try:
import click
except ImportError:
sys.stderr.write('It seems python-dotenv is not installed with cli option. \n'
'Run pip install "python-dotenv[cli]" to fix this.')
sys.exit(1)
from .main import dotenv_values, set_key, unset_key
from .version import __version__
def enumerate_env():
"""
Return a path for the ${pwd}/.env file.
If pwd does not exist, return None.
"""
try:
cwd = os.getcwd()
except FileNotFoundError:
return None
path = os.path.join(cwd, '.env')
return path
@click.group()
@click.option('-f', '--file', default=enumerate_env(),
type=click.Path(file_okay=True),
help="Location of the .env file, defaults to .env file in current working directory.")
@click.option('-q', '--quote', default='always',
type=click.Choice(['always', 'never', 'auto']),
help="Whether to quote or not the variable values. Default mode is always. This does not affect parsing.")
@click.option('-e', '--export', default=False,
type=click.BOOL,
help="Whether to write the dot file as an executable bash script.")
@click.version_option(version=__version__)
@click.pass_context
def cli(ctx: click.Context, file: Any, quote: Any, export: Any) -> None:
"""This script is used to set, get or unset values from a .env file."""
ctx.obj = {'QUOTE': quote, 'EXPORT': export, 'FILE': file}
@contextmanager
def stream_file(path: os.PathLike) -> Iterator[IO[str]]:
"""
Open a file and yield the corresponding (decoded) stream.
Exits with error code 2 if the file cannot be opened.
"""
try:
with open(path) as stream:
yield stream
except OSError as exc:
print(f"Error opening env file: {exc}", file=sys.stderr)
exit(2)
@cli.command()
@click.pass_context
@click.option('--format', default='simple',
type=click.Choice(['simple', 'json', 'shell', 'export']),
help="The format in which to display the list. Default format is simple, "
"which displays name=value without quotes.")
def list(ctx: click.Context, format: bool) -> None:
"""Display all the stored key/value."""
file = ctx.obj['FILE']
with stream_file(file) as stream:
values = dotenv_values(stream=stream)
if format == 'json':
click.echo(json.dumps(values, indent=2, sort_keys=True))
else:
prefix = 'export ' if format == 'export' else ''
for k in sorted(values):
v = values[k]
if v is not None:
if format in ('export', 'shell'):
v = shlex.quote(v)
click.echo(f'{prefix}{k}={v}')
@cli.command()
@click.pass_context
@click.argument('key', required=True)
@click.argument('value', required=True)
def set(ctx: click.Context, key: Any, value: Any) -> None:
"""Store the given key/value."""
file = ctx.obj['FILE']
quote = ctx.obj['QUOTE']
export = ctx.obj['EXPORT']
success, key, value = set_key(file, key, value, quote, export)
if success:
click.echo(f'{key}={value}')
else:
exit(1)
@cli.command()
@click.pass_context
@click.argument('key', required=True)
def get(ctx: click.Context, key: Any) -> None:
"""Retrieve the value for the given key."""
file = ctx.obj['FILE']
with stream_file(file) as stream:
values = dotenv_values(stream=stream)
stored_value = values.get(key)
if stored_value:
click.echo(stored_value)
else:
exit(1)
@cli.command()
@click.pass_context
@click.argument('key', required=True)
def unset(ctx: click.Context, key: Any) -> None:
"""Removes the given key."""
file = ctx.obj['FILE']
quote = ctx.obj['QUOTE']
success, key = unset_key(file, key, quote)
if success:
click.echo(f"Successfully removed {key}")
else:
exit(1)
@cli.command(context_settings={'ignore_unknown_options': True})
@click.pass_context
@click.option(
"--override/--no-override",
default=True,
help="Override variables from the environment file with those from the .env file.",
)
@click.argument('commandline', nargs=-1, type=click.UNPROCESSED)
def run(ctx: click.Context, override: bool, commandline: List[str]) -> None:
"""Run command with environment variables present."""
file = ctx.obj['FILE']
if not os.path.isfile(file):
raise click.BadParameter(
f'Invalid value for \'-f\' "{file}" does not exist.',
ctx=ctx
)
dotenv_as_dict = {
k: v
for (k, v) in dotenv_values(file).items()
if v is not None and (override or k not in os.environ)
}
if not commandline:
click.echo('No command given.')
exit(1)
ret = run_command(commandline, dotenv_as_dict)
exit(ret)
def run_command(command: List[str], env: Dict[str, str]) -> int:
"""Run command in sub process.
Runs the command in a sub process with the variables from `env`
added in the current environment variables.
Parameters
----------
command: List[str]
The command and it's parameters
env: Dict
The additional environment variables
Returns
-------
int
The return code of the command
"""
# copy the current environment variables and add the vales from
# `env`
cmd_env = os.environ.copy()
cmd_env.update(env)
p = Popen(command,
universal_newlines=True,
bufsize=0,
shell=False,
env=cmd_env)
_, _ = p.communicate()
return p.returncode
```
### backend/env/lib/python3.12/site-packages/watchfiles/cli.py
```python
import argparse
import logging
import os
import shlex
import sys
from pathlib import Path
from textwrap import dedent
from typing import Any, Callable, List, Optional, Tuple, Union, cast
from . import Change
from .filters import BaseFilter, DefaultFilter, PythonFilter
from .run import detect_target_type, import_string, run_process
from .version import VERSION
logger = logging.getLogger('watchfiles.cli')
def resolve_path(path_str: str) -> Path:
path = Path(path_str)
if not path.exists():
raise FileNotFoundError(path)
else:
return path.resolve()
def cli(*args_: str) -> None:
"""
Watch one or more directories and execute either a shell command or a python function on file changes.
Example of watching the current directory and calling a python function:
watchfiles foobar.main
Example of watching python files in two local directories and calling a shell command:
watchfiles --filter python 'pytest --lf' src tests
See https://watchfiles.helpmanual.io/cli/ for more information.
"""
args = args_ or sys.argv[1:]
parser = argparse.ArgumentParser(
prog='watchfiles',
description=dedent((cli.__doc__ or '').strip('\n')),
formatter_class=argparse.RawTextHelpFormatter,
)
parser.add_argument('target', help='Command or dotted function path to run')
parser.add_argument(
'paths', nargs='*', default='.', help='Filesystem paths to watch, defaults to current directory'
)
parser.add_argument(
'--ignore-paths',
nargs='?',
type=str,
help=(
'Specify directories to ignore, '
'to ignore multiple paths use a comma as separator, e.g. "env" or "env,node_modules"'
),
)
parser.add_argument(
'--target-type',
nargs='?',
type=str,
default='auto',
choices=['command', 'function', 'auto'],
help=(
'Whether the target should be intercepted as a shell command or a python function, '
'defaults to "auto" which infers the target type from the target string'
),
)
parser.add_argument(
'--filter',
nargs='?',
type=str,
default='default',
help=(
'Which files to watch, defaults to "default" which uses the "DefaultFilter", '
'"python" uses the "PythonFilter", "all" uses no filter, '
'any other value is interpreted as a python function/class path which is imported'
),
)
parser.add_argument(
'--args',
nargs='?',
type=str,
help='Arguments to set on sys.argv before calling target function, used only if the target is a function',
)
parser.add_argument('--verbose', action='store_true', help='Set log level to "debug", wins over `--verbosity`')
parser.add_argument(
'--non-recursive', action='store_true', help='Do not watch for changes in sub-directories recursively'
)
parser.add_argument(
'--verbosity',
nargs='?',
type=str,
default='info',
choices=['warning', 'info', 'debug'],
help='Log level, defaults to "info"',
)
parser.add_argument(
'--sigint-timeout',
nargs='?',
type=int,
default=5,
help='How long to wait for the sigint timeout before sending sigkill.',
)
parser.add_argument(
'--grace-period',
nargs='?',
type=float,
default=0,
help='Number of seconds after the process is started before watching for changes.',
)
parser.add_argument(
'--sigkill-timeout',
nargs='?',
type=int,
default=1,
help='How long to wait for the sigkill timeout before issuing a timeout exception.',
)
parser.add_argument(
'--ignore-permission-denied',
action='store_true',
help='Ignore permission denied errors while watching files and directories.',
)
parser.add_argument('--version', '-V', action='version', version=f'%(prog)s v{VERSION}')
arg_namespace = parser.parse_args(args)
if arg_namespace.verbose:
log_level = logging.DEBUG
else:
log_level = getattr(logging, arg_namespace.verbosity.upper())
hdlr = logging.StreamHandler()
hdlr.setLevel(log_level)
hdlr.setFormatter(logging.Formatter(fmt='[%(asctime)s] %(message)s', datefmt='%H:%M:%S'))
wg_logger = logging.getLogger('watchfiles')
wg_logger.addHandler(hdlr)
wg_logger.setLevel(log_level)
if arg_namespace.target_type == 'auto':
target_type = detect_target_type(arg_namespace.target)
else:
target_type = arg_namespace.target_type
if target_type == 'function':
logger.debug('target_type=function, attempting import of "%s"', arg_namespace.target)
import_exit(arg_namespace.target)
if arg_namespace.args:
sys.argv = [arg_namespace.target] + shlex.split(arg_namespace.args)
elif arg_namespace.args:
logger.warning('--args is only used when the target is a function')
try:
paths = [resolve_path(p) for p in arg_namespace.paths]
except FileNotFoundError as e:
print(f'path "{e}" does not exist', file=sys.stderr)
sys.exit(1)
watch_filter, watch_filter_str = build_filter(arg_namespace.filter, arg_namespace.ignore_paths)
logger.info(
'watchfiles v%s đź‘€ path=%s target="%s" (%s) filter=%s...',
VERSION,
', '.join(f'"{p}"' for p in paths),
arg_namespace.target,
target_type,
watch_filter_str,
)
run_process(
*paths,
target=arg_namespace.target,
target_type=target_type,
watch_filter=watch_filter,
debug=log_level == logging.DEBUG,
sigint_timeout=arg_namespace.sigint_timeout,
sigkill_timeout=arg_namespace.sigkill_timeout,
recursive=not arg_namespace.non_recursive,
ignore_permission_deni
[truncated — 1705 more characters]
```
[3527 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]