generated from alphane/template
rewrote dev command to python, and added prod
This commit is contained in:
@@ -1,295 +1,253 @@
|
||||
#!/usr/bin/env ruby
|
||||
#!/usr/bin/env python3
|
||||
|
||||
class DevHelper
|
||||
# Support dashes in command names
|
||||
COMMAND_TO_METHOD = {
|
||||
"ts-node" => :ts_node,
|
||||
"check-types" => :check_types,
|
||||
"bash-completions" => :bash_completions,
|
||||
"plantuml-to-png" => :plantuml_to_png,
|
||||
}
|
||||
METHOD_TO_COMMAND = COMMAND_TO_METHOD.invert
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
REPLACE_PROCESS = "replace_process"
|
||||
WAIT_FOR_PROCESS = "wait_for_process"
|
||||
# Support dashes in command names
|
||||
COMMAND_TO_METHOD = {
|
||||
"ts-node": "ts_node",
|
||||
"check-types": "check_types",
|
||||
"bash-completions": "bash_completions",
|
||||
"plantuml-to-png": "plantuml_to_png",
|
||||
}
|
||||
METHOD_TO_COMMAND = {method: command for command, method in COMMAND_TO_METHOD.items()}
|
||||
|
||||
# External Interface
|
||||
def self.call(*args)
|
||||
new.call(*args)
|
||||
end
|
||||
WAIT_FOR_PROCESS = "wait_for_process"
|
||||
|
||||
# Core logic
|
||||
def call(*args, **kwargs)
|
||||
command = args[0]
|
||||
method = COMMAND_TO_METHOD.fetch(command, command)
|
||||
if args.length.positive? && respond_to?(method)
|
||||
public_send(method, *args.drop(1), **kwargs)
|
||||
else
|
||||
compose(*args, **kwargs)
|
||||
end
|
||||
end
|
||||
|
||||
def compose(*args, **kwargs)
|
||||
command = compose_command(*args, **kwargs)
|
||||
puts "Running: #{command}" unless kwargs[:slient]
|
||||
class DevHelper:
|
||||
def __init__(self):
|
||||
self.last_exit_status = 0
|
||||
|
||||
case kwargs[:execution_mode]
|
||||
when WAIT_FOR_PROCESS
|
||||
wait_for_process_with_logging(command)
|
||||
else
|
||||
exec(command)
|
||||
end
|
||||
end
|
||||
# Core logic
|
||||
def call(self, *args, **kwargs):
|
||||
if not args:
|
||||
return self.compose(*args, **kwargs)
|
||||
|
||||
# Primary command wrappers
|
||||
def build(*args, **kwargs)
|
||||
compose(%w[build], *args, **kwargs)
|
||||
end
|
||||
command, rest = args[0], args[1:]
|
||||
method_name = COMMAND_TO_METHOD.get(command, command)
|
||||
method = getattr(self, method_name, None)
|
||||
if callable(method) and not method_name.startswith("_") and method_name != "call":
|
||||
return method(*rest, **kwargs)
|
||||
return self.compose(*args, **kwargs)
|
||||
|
||||
def compile(*args, **kwargs)
|
||||
run(*%w[api npm run build], execution_mode: WAIT_FOR_PROCESS)
|
||||
exit($?.exitstatus) unless $?.success?
|
||||
run(*%w[web npm run build])
|
||||
end
|
||||
def compose(self, *args, **kwargs):
|
||||
command = self._compose_command(*args, **kwargs)
|
||||
if not kwargs.get("silent"):
|
||||
print(f"Running: {command}")
|
||||
|
||||
def up(*args, **kwargs)
|
||||
compose(*%w[up --remove-orphans], *args, **kwargs)
|
||||
end
|
||||
if kwargs.get("execution_mode") == WAIT_FOR_PROCESS:
|
||||
self._wait_for_process_with_logging(command)
|
||||
else:
|
||||
os.execv("/bin/sh", ["/bin/sh", "-c", command])
|
||||
|
||||
def down(*args, **kwargs)
|
||||
compose(*%w[down --remove-orphans], *args, **kwargs)
|
||||
end
|
||||
# Primary command wrappers
|
||||
def build(self, *args, **kwargs):
|
||||
self.compose("build", *args, **kwargs)
|
||||
|
||||
def logs(*args, **kwargs)
|
||||
compose(*%w[logs -f], *args, **kwargs)
|
||||
end
|
||||
def compile(self, *args, **kwargs):
|
||||
self.run("api", "npm", "run", "build", execution_mode=WAIT_FOR_PROCESS)
|
||||
if self.last_exit_status != 0:
|
||||
sys.exit(self.last_exit_status)
|
||||
self.run("web", "npm", "run", "build")
|
||||
|
||||
def run(*args, **kwargs)
|
||||
compose(*%w[run --rm], *args, **kwargs)
|
||||
end
|
||||
def up(self, *args, **kwargs):
|
||||
self.compose("up", "--remove-orphans", *args, **kwargs)
|
||||
|
||||
def ps(*args, **kwargs)
|
||||
compose(*%w[ps], *args, **kwargs)
|
||||
end
|
||||
def down(self, *args, **kwargs):
|
||||
self.compose("down", "--remove-orphans", *args, **kwargs)
|
||||
|
||||
# Custom helpers
|
||||
def api(*args, **kwargs)
|
||||
run(*%w[api], *args, **kwargs)
|
||||
end
|
||||
def logs(self, *args, **kwargs):
|
||||
self.compose("logs", "-f", *args, **kwargs)
|
||||
|
||||
def web(*args, **kwargs)
|
||||
run(*%w[web], *args, **kwargs)
|
||||
end
|
||||
def run(self, *args, **kwargs):
|
||||
self.compose("run", "--rm", *args, **kwargs)
|
||||
|
||||
def check_types(*args, **kwargs)
|
||||
run(*%w[api npm run check-types], *args, **kwargs)
|
||||
end
|
||||
def ps(self, *args, **kwargs):
|
||||
self.compose("ps", *args, **kwargs)
|
||||
|
||||
def test(*args, **kwargs)
|
||||
service = args[0]
|
||||
if service == "api"
|
||||
test_api(*args.drop(1), **kwargs)
|
||||
elsif service == "web"
|
||||
test_web(*args.drop(1), **kwargs)
|
||||
else
|
||||
test_api(*args, **kwargs)
|
||||
end
|
||||
end
|
||||
# Custom helpers
|
||||
def api(self, *args, **kwargs):
|
||||
self.run("api", *args, **kwargs)
|
||||
|
||||
def test_api(*args, **kwargs)
|
||||
reformat_project_relative_path_filter_for_vitest!(args, "api/")
|
||||
run(*%w[test_api npm run test], *args, **kwargs)
|
||||
end
|
||||
def web(self, *args, **kwargs):
|
||||
self.run("web", *args, **kwargs)
|
||||
|
||||
def test_web(*args, **kwargs)
|
||||
reformat_project_relative_path_filter_for_vitest!(args, "web/")
|
||||
run(*%w[test_web npm run test], *args, **kwargs)
|
||||
end
|
||||
def check_types(self, *args, **kwargs):
|
||||
self.run("api", "npm", "run", "check-types", *args, **kwargs)
|
||||
|
||||
def sqlcmd(*args, **kwargs)
|
||||
db_host = ENV.fetch('DB_HOST', 'localhost')
|
||||
db_user = ENV.fetch('DB_USER', 'sa')
|
||||
db_pass = ENV.fetch('DB_PASS', '1m5ecure!')
|
||||
db_name = ENV.fetch('DB_NAME', 'YHSI')
|
||||
compose(
|
||||
*%w[exec db /opt/mssql-tools/bin/sqlcmd],
|
||||
*%W[-U #{db_user}],
|
||||
*%W[-P #{db_pass}],
|
||||
*%W[-H #{db_host}],
|
||||
*%W[-d #{db_name}],
|
||||
'-I', # enable quoted identifiers, e.g. "table"."column"
|
||||
*args,
|
||||
**kwargs
|
||||
)
|
||||
end
|
||||
def test(self, *args, **kwargs):
|
||||
service = args[0] if args else None
|
||||
if service == "api":
|
||||
self.test_api(*args[1:], **kwargs)
|
||||
elif service == "web":
|
||||
self.test_web(*args[1:], **kwargs)
|
||||
else:
|
||||
self.test_api(*args, **kwargs)
|
||||
|
||||
def db(*args, **kwargs)
|
||||
compose(*%w[exec db], *args, **kwargs)
|
||||
end
|
||||
def test_api(self, *args, **kwargs):
|
||||
args = self._reformat_project_relative_path_filter_for_vitest(list(args), "api/")
|
||||
self.run("test_api", "npm", "run", "test", *args, **kwargs)
|
||||
|
||||
def debug
|
||||
api_container_id = container_id("api")
|
||||
puts "Waiting for breakpoint to trigger..."
|
||||
puts "'ctrl-c' to exit."
|
||||
command = "docker attach --detach-keys ctrl-c #{api_container_id}"
|
||||
puts "Running: #{command}"
|
||||
exec(command)
|
||||
exit 0
|
||||
end
|
||||
def test_web(self, *args, **kwargs):
|
||||
args = self._reformat_project_relative_path_filter_for_vitest(list(args), "web/")
|
||||
self.run("test_web", "npm", "run", "test", *args, **kwargs)
|
||||
|
||||
def npm(*args, **kwargs)
|
||||
run(*%w[api npm], *args, **kwargs)
|
||||
end
|
||||
def sqlcmd(self, *args, **kwargs):
|
||||
db_host = os.environ.get("DB_HOST", "localhost")
|
||||
db_user = os.environ.get("DB_USER", "sa")
|
||||
db_pass = os.environ.get("DB_PASS", "1m5ecure!")
|
||||
db_name = os.environ.get("DB_NAME", "YHSI")
|
||||
self.compose(
|
||||
"exec",
|
||||
"db",
|
||||
"/opt/mssql-tools/bin/sqlcmd",
|
||||
"-U",
|
||||
db_user,
|
||||
"-P",
|
||||
db_pass,
|
||||
"-H",
|
||||
db_host,
|
||||
"-d",
|
||||
db_name,
|
||||
"-I", # enable quoted identifiers, e.g. "table"."column"
|
||||
*args,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def ts_node(*args, **kwargs)
|
||||
run(*%w[api npm run ts-node], *args, **kwargs)
|
||||
end
|
||||
def db(self, *args, **kwargs):
|
||||
self.compose("exec", "db", *args, **kwargs)
|
||||
|
||||
def knex(*args, **kwargs)
|
||||
if RUBY_PLATFORM =~ /linux/
|
||||
run(*%w[api npm run knex], *args, execution_mode: WAIT_FOR_PROCESS, **kwargs)
|
||||
def debug(self, *args, **kwargs):
|
||||
container_id = self._container_id("api")
|
||||
print("Waiting for breakpoint to trigger...")
|
||||
print("'ctrl-c' to exit.")
|
||||
command = f"docker attach --detach-keys ctrl-c {container_id}"
|
||||
print(f"Running: {command}")
|
||||
os.execv("/bin/sh", ["/bin/sh", "-c", command])
|
||||
|
||||
file_or_directory = "#{project_root}/api/src/db/migrations"
|
||||
exit(0) unless take_over_needed?(file_or_directory)
|
||||
def npm(self, *args, **kwargs):
|
||||
self.run("api", "npm", *args, **kwargs)
|
||||
|
||||
ownit file_or_directory
|
||||
else
|
||||
run(*%w[api npm run knex], *args, **kwargs)
|
||||
end
|
||||
end
|
||||
def ts_node(self, *args, **kwargs):
|
||||
self.run("api", "npm", "run", "ts-node", *args, **kwargs)
|
||||
|
||||
def migrate(*args, **kwargs)
|
||||
action = args[0]
|
||||
knex("migrate:#{action}", *args.drop(1), **kwargs)
|
||||
end
|
||||
def knex(self, *args, **kwargs):
|
||||
if sys.platform.startswith("linux"):
|
||||
self.run("api", "npm", "run", "knex", *args, execution_mode=WAIT_FOR_PROCESS, **kwargs)
|
||||
|
||||
def seed(*args, **kwargs)
|
||||
action = args[0]
|
||||
knex("seed:#{action}", *args.drop(1), **kwargs)
|
||||
end
|
||||
file_or_directory = os.path.join(self._project_root(), "api/src/db/migrations")
|
||||
if not self._take_over_needed(file_or_directory):
|
||||
sys.exit(0)
|
||||
|
||||
def ownit(*args, **kwargs)
|
||||
file_or_directory = args[0]
|
||||
raise ScriptError, "Must provide a file or directory path." if file_or_directory.nil?
|
||||
self.ownit(file_or_directory)
|
||||
else:
|
||||
self.run("api", "npm", "run", "knex", *args, **kwargs)
|
||||
|
||||
if RUBY_PLATFORM =~ /linux/
|
||||
puts "Take ownership of the file or directory? #{file_or_directory}"
|
||||
exec("sudo chown -R #{user_id}:#{group_id} #{file_or_directory}")
|
||||
else
|
||||
raise NotImplementedError, "Not implement for platform #{RUBY_PLATFORM}"
|
||||
end
|
||||
end
|
||||
def migrate(self, *args, **kwargs):
|
||||
action = args[0] if args else None
|
||||
self.knex(f"migrate:{action}", *args[1:], **kwargs)
|
||||
|
||||
def plantuml_to_png(*args, **kwargs)
|
||||
file_path = args.pop
|
||||
raise ScriptError, "Must provide a file path." if file_path.nil?
|
||||
def seed(self, *args, **kwargs):
|
||||
action = args[0] if args else None
|
||||
self.knex(f"seed:{action}", *args[1:], **kwargs)
|
||||
|
||||
png_path = file_path.gsub(/\.(wsd|pu|puml|plantuml|uml)$/, ".png")
|
||||
def ownit(self, *args, **kwargs):
|
||||
file_or_directory = args[0] if args else None
|
||||
if file_or_directory is None:
|
||||
raise ValueError("Must provide a file or directory path.")
|
||||
|
||||
command = <<~BASH
|
||||
curl #{args.join(" ")} \
|
||||
--data-binary @'#{file_path}' \
|
||||
http://localhost:9999/png > '#{png_path}'
|
||||
BASH
|
||||
if sys.platform.startswith("linux"):
|
||||
print(f"Take ownership of the file or directory? {file_or_directory}")
|
||||
command = f"sudo chown -R {self._user_id()}:{self._group_id()} {file_or_directory}"
|
||||
os.execv("/bin/sh", ["/bin/sh", "-c", command])
|
||||
else:
|
||||
raise NotImplementedError(f"Not implemented for platform {sys.platform}")
|
||||
|
||||
puts "Running: #{command}"
|
||||
exec(command)
|
||||
end
|
||||
def plantuml_to_png(self, *args, **kwargs):
|
||||
args = list(args)
|
||||
if not args:
|
||||
raise ValueError("Must provide a file path.")
|
||||
file_path = args.pop()
|
||||
|
||||
def bash_completions
|
||||
completions =
|
||||
public_methods(false)
|
||||
.reject { |word| %i[call].include?(word) }
|
||||
.map { |word| METHOD_TO_COMMAND.fetch(word, word) }
|
||||
puts completions
|
||||
end
|
||||
png_path = re.sub(r"\.(wsd|pu|puml|plantuml|uml)$", ".png", file_path)
|
||||
|
||||
private
|
||||
command = f"curl {' '.join(args)} --data-binary @'{file_path}' http://localhost:9999/png > '{png_path}'"
|
||||
|
||||
def wait_for_process_with_logging(command)
|
||||
IO.popen("#{command} 2>&1") do |io|
|
||||
until io.eof?
|
||||
line = io.gets
|
||||
puts line
|
||||
end
|
||||
end
|
||||
end
|
||||
print(f"Running: {command}")
|
||||
os.execv("/bin/sh", ["/bin/sh", "-c", command])
|
||||
|
||||
def container_id(container_name, *args, **kwargs)
|
||||
command = compose_command(*%w[ps -q], container_name, *args, **kwargs)
|
||||
puts "Running: #{command}"
|
||||
id_of_container = `#{command}`.chomp
|
||||
puts "Container id is: #{id_of_container}"
|
||||
id_of_container
|
||||
end
|
||||
def bash_completions(self, *args, **kwargs):
|
||||
completions = sorted(
|
||||
METHOD_TO_COMMAND.get(name, name)
|
||||
for name in vars(DevHelper)
|
||||
if not name.startswith("_") and name != "call" and callable(getattr(DevHelper, name))
|
||||
)
|
||||
print(" ".join(completions))
|
||||
|
||||
def service_running?(container_name)
|
||||
ps(*%w[-q --status=running], execution_mode: WAIT_FOR_PROCESS, slient: true) != ""
|
||||
end
|
||||
# Private helpers
|
||||
def _wait_for_process_with_logging(self, command):
|
||||
process = subprocess.Popen(
|
||||
["/bin/sh", "-c", f"{command} 2>&1"],
|
||||
stdout=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
for line in process.stdout:
|
||||
print(line, end="")
|
||||
process.wait()
|
||||
self.last_exit_status = process.returncode
|
||||
|
||||
def compose_command(*args, **kwargs)
|
||||
environment = kwargs.fetch(:environment, "development")
|
||||
"cd #{project_root} && docker compose -f docker-compose.#{environment}.yml #{args.join(" ")}"
|
||||
end
|
||||
def _container_id(self, container_name, *args, **kwargs):
|
||||
command = self._compose_command("ps", "-q", container_name, *args, **kwargs)
|
||||
print(f"Running: {command}")
|
||||
result = subprocess.run(["/bin/sh", "-c", command], stdout=subprocess.PIPE, text=True)
|
||||
container_id = result.stdout.strip()
|
||||
print(f"Container id is: {container_id}")
|
||||
return container_id
|
||||
|
||||
def project_root
|
||||
@project_root ||= File.absolute_path("#{__dir__}/..")
|
||||
end
|
||||
def _service_running(self, container_name):
|
||||
self.ps("-q", "--status=running", execution_mode=WAIT_FOR_PROCESS, silent=True)
|
||||
return self.last_exit_status == 0
|
||||
|
||||
def take_over_needed?(file_or_directory)
|
||||
files_owned_by_others =
|
||||
system("find #{file_or_directory} -not -user #{user_id} -print -quit | grep -q .")
|
||||
files_owned_by_others
|
||||
end
|
||||
def _compose_command(self, *args, **kwargs):
|
||||
environment = kwargs.get("environment", "development")
|
||||
return f"cd {self._project_root()} && docker compose -f docker-compose.{environment}.yml {' '.join(args)}"
|
||||
|
||||
def user_id
|
||||
unless RUBY_PLATFORM =~ /linux/
|
||||
raise NotImplementedError, "Not implement for platform #{RUBY_PLATFORM}"
|
||||
end
|
||||
def _project_root(self):
|
||||
return os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
`id -u`.strip
|
||||
end
|
||||
def _take_over_needed(self, file_or_directory):
|
||||
result = subprocess.run(
|
||||
["/bin/sh", "-c", f"find {file_or_directory} -not -user {self._user_id()} -print -quit | grep -q ."],
|
||||
)
|
||||
return result.returncode == 0
|
||||
|
||||
def group_id
|
||||
unless RUBY_PLATFORM =~ /linux/
|
||||
raise NotImplementedError, "Not implement for platform #{RUBY_PLATFORM}"
|
||||
end
|
||||
def _user_id(self):
|
||||
if not sys.platform.startswith("linux"):
|
||||
raise NotImplementedError(f"Not implemented for platform {sys.platform}")
|
||||
return subprocess.run(["id", "-u"], stdout=subprocess.PIPE, text=True).stdout.strip()
|
||||
|
||||
`id -g`.strip
|
||||
end
|
||||
def _group_id(self):
|
||||
if not sys.platform.startswith("linux"):
|
||||
raise NotImplementedError(f"Not implemented for platform {sys.platform}")
|
||||
return subprocess.run(["id", "-g"], stdout=subprocess.PIPE, text=True).stdout.strip()
|
||||
|
||||
def reformat_project_relative_path_filter_for_vitest!(args, prefix)
|
||||
if args.length.positive? && args[0].start_with?(prefix)
|
||||
src_path_prefix = "#{prefix}src/"
|
||||
test_path_regex = Regexp.escape(prefix)
|
||||
src_path_regex = Regexp.escape(src_path_prefix)
|
||||
def _reformat_project_relative_path_filter_for_vitest(self, args, prefix):
|
||||
if args and args[0].startswith(prefix):
|
||||
src_path_prefix = f"{prefix}src/"
|
||||
|
||||
if args[0].start_with?(src_path_prefix)
|
||||
# TODO: handle other file types
|
||||
args[0] = args[0].gsub(/^#{src_path_regex}/, "tests/").gsub(/\.ts$/, ".test.ts")
|
||||
else
|
||||
args[0] = args[0].gsub(/^#{test_path_regex}/, "")
|
||||
end
|
||||
if args[0].startswith(src_path_prefix):
|
||||
# TODO: handle other file types
|
||||
args[0] = re.sub(r"\.ts$", ".test.ts", re.sub(f"^{re.escape(src_path_prefix)}", "tests/", args[0]))
|
||||
else:
|
||||
args[0] = re.sub(f"^{re.escape(prefix)}", "", args[0])
|
||||
|
||||
puts "Reformatted path filter from project relative to service relative for vitest."
|
||||
end
|
||||
end
|
||||
end
|
||||
print("Reformatted path filter from project relative to service relative for vitest.")
|
||||
|
||||
# Only execute main function when file is executed
|
||||
DevHelper.call(*ARGV) if $PROGRAM_NAME == __FILE__
|
||||
return args
|
||||
|
||||
## Dev completions
|
||||
# https://iridakos.com/programming/2018/03/01/bash-programmable-completion-tutorial
|
||||
# _dev_completions () {
|
||||
# local dev_command_path="$(which dev)"
|
||||
# local dev_function_names
|
||||
# dev_function_names="$(ruby "$dev_command_path" bash_completions)"
|
||||
# # COMP_WORDS: an array of all the words typed after the name of the program the compspec belongs to
|
||||
# # COMP_CWORD: an index of the COMP_WORDS array pointing to the word the current cursor is at - in other words, the index of the word the cursor was when the tab key was pressed
|
||||
# # COMP_LINE: the current command line
|
||||
# COMPREPLY=($(compgen -W "$dev_function_names" "${COMP_WORDS[$COMP_CWORD]}"))
|
||||
# }
|
||||
|
||||
# complete -F _dev_completions dev
|
||||
# complete -W "allow" direnv
|
||||
if __name__ == "__main__":
|
||||
DevHelper().call(*sys.argv[1:])
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
# Support dashes in command names
|
||||
COMMAND_TO_METHOD = {}
|
||||
METHOD_TO_COMMAND = {method: command for command, method in COMMAND_TO_METHOD.items()}
|
||||
|
||||
WAIT_FOR_PROCESS = "wait_for_process"
|
||||
|
||||
|
||||
class ProdHelper:
|
||||
def __init__(self):
|
||||
self.last_exit_status = 0
|
||||
|
||||
# Core logic
|
||||
def call(self, *args, **kwargs):
|
||||
if not args:
|
||||
return self.compose(*args, **kwargs)
|
||||
|
||||
command, rest = args[0], args[1:]
|
||||
method_name = COMMAND_TO_METHOD.get(command, command)
|
||||
method = getattr(self, method_name, None)
|
||||
if callable(method) and not method_name.startswith("_") and method_name != "call":
|
||||
return method(*rest, **kwargs)
|
||||
return self.compose(*args, **kwargs)
|
||||
|
||||
def compose(self, *args, **kwargs):
|
||||
command = self._compose_command(*args, **kwargs)
|
||||
if not kwargs.get("silent"):
|
||||
print(f"Running: {command}")
|
||||
|
||||
if kwargs.get("execution_mode") == WAIT_FOR_PROCESS:
|
||||
self._wait_for_process_with_logging(command)
|
||||
else:
|
||||
os.execv("/bin/sh", ["/bin/sh", "-c", command])
|
||||
|
||||
# Primary command wrappers
|
||||
def up(self, *args, **kwargs):
|
||||
self.compose("up", "--remove-orphans", *args, **kwargs)
|
||||
|
||||
def down(self, *args, **kwargs):
|
||||
self.compose("down", "--remove-orphans", *args, **kwargs)
|
||||
|
||||
def logs(self, *args, **kwargs):
|
||||
self.compose("logs", "-f", *args, **kwargs)
|
||||
|
||||
def ps(self, *args, **kwargs):
|
||||
self.compose("ps", *args, **kwargs)
|
||||
|
||||
def bash_completions(self, *args, **kwargs):
|
||||
completions = sorted(
|
||||
METHOD_TO_COMMAND.get(name, name)
|
||||
for name in vars(ProdHelper)
|
||||
if not name.startswith("_") and name != "call" and callable(getattr(ProdHelper, name))
|
||||
)
|
||||
print(" ".join(completions))
|
||||
|
||||
# Private helpers
|
||||
def _wait_for_process_with_logging(self, command):
|
||||
process = subprocess.Popen(
|
||||
["/bin/sh", "-c", f"{command} 2>&1"],
|
||||
stdout=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
for line in process.stdout:
|
||||
print(line, end="")
|
||||
process.wait()
|
||||
self.last_exit_status = process.returncode
|
||||
|
||||
def _compose_command(self, *args, **kwargs):
|
||||
environment = kwargs.get("environment", "production")
|
||||
return f"cd {self._project_root()} && docker compose -f docker-compose.{environment}.yml {' '.join(args)}"
|
||||
|
||||
def _project_root(self):
|
||||
return os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
ProdHelper().call(*sys.argv[1:])
|
||||
Reference in New Issue
Block a user