Coverage for .tox/coverage/lib/python3.13/site-packages/wuttjamaican/install.py: 100%
209 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-27 13:03 -0500
1# -*- coding: utf-8; -*-
2################################################################################
3#
4# WuttJamaican -- Base package for Wutta Framework
5# Copyright © 2023-2026 Lance Edgar
6#
7# This file is part of Wutta Framework.
8#
9# Wutta Framework is free software: you can redistribute it and/or modify it
10# under the terms of the GNU General Public License as published by the Free
11# Software Foundation, either version 3 of the License, or (at your option) any
12# later version.
13#
14# Wutta Framework is distributed in the hope that it will be useful, but
15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
17# more details.
18#
19# You should have received a copy of the GNU General Public License along with
20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
21#
22################################################################################
23"""
24Install Handler
25"""
27import os
28import stat
29import subprocess
30import sys
32import rich
33from mako.lookup import TemplateLookup
35from wuttjamaican.app import GenericHandler
38class InstallHandler(GenericHandler): # pylint: disable=too-many-public-methods
39 """
40 Base class and default implementation for the :term:`install
41 handler`.
43 The installer runs interactively via command line. If
44 installation completes okay the exit code is 0, but if not:
46 * exit code 1 indicates user canceled
47 * exit code 2 indicates sanity check failed
48 * other codes possible if errors occur
50 Usually an app will define e.g. ``poser install`` command which
51 would invoke the install handler's :meth:`run()` method::
53 app = config.get_app()
54 install = app.get_install_handler(pkg_name="poser")
55 install.run()
57 Note that these first 4 attributes may be specified via
58 constructor kwargs:
60 .. attribute:: pkg_name
62 Python package name for the app, e.g. ``poser``.
64 .. attribute:: app_title
66 Display title for the app, e.g. "Poser".
68 .. attribute:: pypi_name
70 Package distribution name, e.g. for PyPI. If not specified one
71 will be guessed.
73 .. attribute:: egg_name
75 Egg name for the app. If not specified one will be guessed.
77 """
79 pkg_name = "poser"
80 app_title = None
81 pypi_name = None
82 egg_name = None
83 schema_installed = False
85 # nb. we prompt the user for this, unless attr already has value
86 wants_continuum = None
88 template_paths = ["wuttjamaican:templates/install"]
90 def __init__(self, config, **kwargs):
91 super().__init__(config)
93 # nb. caller may specify pkg_name etc.
94 self.__dict__.update(kwargs)
96 # some package names we can generate by default
97 if not self.app_title:
98 self.app_title = self.pkg_name
99 if not self.pypi_name:
100 self.pypi_name = self.app_title
101 if not self.egg_name:
102 self.egg_name = self.pypi_name.replace("-", "_")
104 paths = [self.app.resource_path(p) for p in self.template_paths]
106 try:
107 paths.insert(
108 0, self.app.resource_path(f"{self.pkg_name}:templates/install")
109 )
110 except (TypeError, ModuleNotFoundError):
111 pass
113 self.templates = TemplateLookup(directories=paths)
115 # nb. this will eventually hold all context from user prompts etc.
116 self.final_context = None
118 def run(self):
119 """
120 Run the interactive command-line installer.
122 This does the following:
124 * check for ``prompt_toolkit`` and maybe ask to install it
125 * call :meth:`show_welcome()`
126 * call :meth:`sanity_check()`
127 * call :meth:`do_install_steps()`
128 * call :meth:`show_goodbye()`
130 Although if a problem is encountered then not all calls may
131 happen.
132 """
133 self.require_prompt_toolkit()
134 self.show_welcome()
135 self.sanity_check()
136 self.schema_installed = False
137 self.do_install_steps()
138 self.show_goodbye()
140 def show_welcome(self):
141 """
142 Show the intro/welcome message, and prompt user to begin the
143 install.
145 This is normally called by :meth:`run()`.
146 """
147 self.rprint(f"\n\t[blue]Welcome to {self.app.get_title()}![/blue]")
148 self.rprint("\n\tThis tool will install and configure the app.")
149 self.rprint(
150 "\n\t[bold italic]NB. You should already have created "
151 "the database in PostgreSQL or MySQL.[/bold italic]"
152 )
154 # shall we continue?
155 if not self.prompt_bool("continue?", True):
156 self.rprint()
157 sys.exit(1)
159 def sanity_check(self):
160 """
161 Perform various sanity checks before doing the install. If
162 any problem is found the installer should exit with code 2.
164 This is normally called by :meth:`run()`.
166 The default logic here just calls :meth:`check_appdir()`.
167 """
168 self.check_appdir()
170 def check_appdir(self):
171 """
172 Check if the :term:`app dir` already exists; exit with code 2
173 if so.
175 This is normally called from :meth:`sanity_check()`.
176 """
177 # appdir must not yet exist
178 appdir = os.path.join(sys.prefix, "app")
179 if os.path.exists(appdir):
180 self.rprint(f"\n\t[bold red]appdir already exists:[/bold red] {appdir}\n")
181 sys.exit(2)
183 def do_install_steps(self):
184 """
185 Perform the real installation steps.
187 This method is called by :meth:`run()` and does the following:
189 * call :meth:`prompt_user_for_context()` to collect DB info etc.
190 * call :meth:`make_template_context()` to use when generating output
191 * call :meth:`make_appdir()` to create app dir with config files
192 * call :meth:`install_db_schema()` to (optionally) create tables in DB
193 """
194 # prompt user / get context
195 context = self.prompt_user_for_context()
196 context = self.make_template_context(**context)
198 # keep a reference to this in case subsequent logic needs it
199 self.final_context = context
201 # make the appdir
202 self.make_appdir(context)
204 # install db schema if user likes
205 self.schema_installed = self.install_db_schema(context["db_url"])
207 def prompt_user_for_context(self):
208 """
209 This is responsible for initial user prompts.
211 This happens early in the install, so this method can verify
212 the info, e.g. test the DB connection, but should not write
213 any files as the app dir may not exist yet.
215 Default logic calls :meth:`get_db_url()` for the DB
216 connection, then may ask about Wutta-Continuum data
217 versioning. (The latter is skipped if the package is
218 missing.)
220 Subclass should override this method if they need different
221 prompting logic. The return value should always include at
222 least these 2 items:
224 * ``db_url`` - URL for the DB connection
225 * ``wants_continuum`` - whether data versioning should be enabled
227 :returns: Dict of template context
228 """
229 # db info
230 db_url = self.get_db_url()
232 # continuum
233 if self.wants_continuum is None:
234 try:
235 import wutta_continuum # pylint: disable=import-outside-toplevel,unused-import
236 except ImportError:
237 self.wants_continuum = False
238 else:
239 self.wants_continuum = self.prompt_bool(
240 "use continuum for data versioning?", default=True
241 )
243 return {"db_url": db_url, "wants_continuum": self.wants_continuum}
245 def get_db_url(self):
246 """
247 This must return the DB engine URL.
249 Default logic will prompt the user for hostname, port, DB name
250 and credentials. It then assembles the URL from those parts.
252 This method will also test the DB connection. If it fails,
253 the install is aborted.
255 This method is normally called by
256 :meth:`prompt_user_for_context()`.
258 :returns: SQLAlchemy engine URL (as object or string)
259 """
260 # get db info/url
261 dbinfo = self.get_dbinfo()
262 db_url = dbinfo.get("db_url")
263 if not db_url:
264 db_url = self.make_db_url(dbinfo)
266 # test db connection
267 self.rprint("\n\ttesting db connection... ", end="")
268 error = self.test_db_connection(db_url)
269 if error:
270 self.rprint("[bold red]cannot connect![/bold red] ..error was:")
271 self.rprint(f"\n{error}")
272 self.rprint("\n\t[bold yellow]aborting mission[/bold yellow]\n")
273 sys.exit(1)
274 self.rprint("[bold green]good[/bold green]")
276 return db_url
278 def get_dbinfo(self): # pylint: disable=missing-function-docstring
279 dbinfo = {}
281 # main info
282 dbinfo["dbtype"] = self.prompt_generic("db type", "postgresql")
283 dbinfo["dbhost"] = self.prompt_generic("db host", "localhost")
284 default_port = "3306" if dbinfo["dbtype"] == "mysql" else "5432"
285 dbinfo["dbport"] = self.prompt_generic("db port", default_port)
286 dbinfo["dbname"] = self.prompt_generic("db name", self.pkg_name)
287 dbinfo["dbuser"] = self.prompt_generic("db user", self.pkg_name)
289 # password
290 dbinfo["dbpass"] = None
291 while not dbinfo["dbpass"]:
292 dbinfo["dbpass"] = self.prompt_generic("db pass", is_password=True)
294 return dbinfo
296 def make_db_url(self, dbinfo): # pylint: disable=empty-docstring
297 """ """
298 from sqlalchemy.engine import URL # pylint: disable=import-outside-toplevel
300 if dbinfo["dbtype"] == "mysql":
301 drivername = "mysql+mysqlconnector"
302 else:
303 drivername = "postgresql+psycopg"
305 return URL.create(
306 drivername=drivername,
307 username=dbinfo["dbuser"],
308 password=dbinfo["dbpass"],
309 host=dbinfo["dbhost"],
310 port=dbinfo["dbport"],
311 database=dbinfo["dbname"],
312 )
314 def test_db_connection(self, url): # pylint: disable=empty-docstring
315 """ """
316 import sqlalchemy as sa # pylint: disable=import-outside-toplevel
318 engine = sa.create_engine(url)
320 # check for random table; does not matter if it exists, we
321 # just need to test interaction and this is a neutral way
322 try:
323 sa.inspect(engine).has_table("whatever")
324 except Exception as error: # pylint: disable=broad-exception-caught
325 return str(error)
326 return None
328 def make_template_context(self, **kwargs):
329 """
330 This must return a dict to be used as global template context
331 when generating output (e.g. config) files.
333 This method is normally called by :meth:`do_install_steps()`.
334 The ``context`` returned is then passed to
335 :meth:`render_mako_template()`.
337 Note these first 2 params are not explicitly listed in the
338 method signature; they are required nonetheless.
340 :param db_url: This must be a string URL for the DB engine.
342 :param wants_continuum: Whether data versioning should be
343 enabled within the config.
345 :param \\**kwargs: Extra template context.
347 :returns: Dict for global template context.
349 The final context dict should include at least:
351 * ``envdir`` - value from :data:`python:sys.prefix`
352 * ``envname`` - "last" dirname from ``sys.prefix``
353 * ``pkg_name`` - value from :attr:`pkg_name`
354 * ``app_title`` - value from :attr:`app_title`
355 * ``pypi_name`` - value from :attr:`pypi_name`
356 * ``egg_name`` - value from :attr:`egg_name`
357 * ``appdir`` - ``app`` folder under ``sys.prefix``
358 * ``db_url`` - value from ``kwargs``
359 * ``wants_continuum`` - value from ``kwargs``
360 """
361 envname = os.path.basename(sys.prefix)
362 appdir = os.path.join(sys.prefix, "app")
364 db_url = kwargs.pop("db_url")
365 if not isinstance(db_url, str):
366 db_url = db_url.render_as_string(hide_password=False)
368 context = {
369 "envdir": sys.prefix,
370 "envname": envname,
371 "pkg_name": self.pkg_name,
372 "app_title": self.app_title,
373 "pypi_name": self.pypi_name,
374 "appdir": appdir,
375 "egg_name": self.egg_name,
376 "db_url": db_url,
377 }
378 context.update(kwargs)
379 return context
381 def make_appdir(self, context, appdir=None):
382 """
383 Create the app folder structure and generate config files.
385 This method is normally called by :meth:`do_install_steps()`.
387 :param context: Template context dict, i.e. from
388 :meth:`make_template_context()`.
390 The default logic will create a structure as follows, assuming
391 ``/venv`` is the path to the virtual environment:
393 .. code-block:: none
395 /venv/
396 └── app/
397 ├── cache/
398 ├── data/
399 ├── log/
400 ├── work/
401 ├── wutta.conf
402 ├── web.conf
403 └── upgrade.sh
405 File templates for this come from
406 ``wuttjamaican:templates/install`` by default.
408 This method calls
409 :meth:`~wuttjamaican.app.AppHandler.make_appdir()` for the
410 basic structure and then :meth:`write_all_config_files()` for
411 the gory details.
412 """
413 # app handler makes appdir proper
414 appdir = appdir or self.app.get_appdir()
415 self.app.make_appdir(appdir)
417 # but then we also generate some files...
418 self.write_all_config_files(appdir, context)
420 self.rprint(f"\n\tappdir created at: [bold green]{appdir}[/bold green]")
422 def write_all_config_files(self, appdir, context):
423 """
424 This method should write all config files within the app dir.
425 It's called from :meth:`make_appdir()`.
427 Subclass can override this for specialized installers.
429 Note that the app dir may or may not be newly-created, when
430 this method is called. Some installers may support a
431 "refresh" of the existing app dir.
433 Default logic (over)writes 3 files:
435 * ``wutta.conf``
436 * ``web.cof``
437 * ``upgrade.sh``
438 """
439 self.write_wutta_conf(appdir, context)
440 self.write_web_conf(appdir, context)
441 self.write_upgrade_sh(appdir, context)
443 def write_wutta_conf(
444 self, appdir, context
445 ): # pylint: disable=missing-function-docstring
446 self.make_config_file(
447 "wutta.conf.mako", os.path.join(appdir, "wutta.conf"), **context
448 )
450 def write_web_conf(
451 self, appdir, context
452 ): # pylint: disable=missing-function-docstring
453 web_context = dict(context)
454 web_context.setdefault("beaker_key", context.get("pkg_name", "poser"))
455 web_context.setdefault("beaker_secret", "TODO_YOU_SHOULD_CHANGE_THIS")
456 web_context.setdefault("pyramid_host", "0.0.0.0")
457 web_context.setdefault("pyramid_port", "9080")
458 self.make_config_file(
459 "web.conf.mako", os.path.join(appdir, "web.conf"), **web_context
460 )
462 def write_upgrade_sh(
463 self, appdir, context
464 ): # pylint: disable=missing-function-docstring
465 template = self.templates.get_template("upgrade.sh.mako")
466 output_path = os.path.join(appdir, "upgrade.sh")
467 self.render_mako_template(template, context, output_path=output_path)
468 os.chmod(
469 output_path,
470 stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH,
471 )
473 def render_mako_template(
474 self,
475 template,
476 context,
477 output_path=None,
478 ):
479 """
480 Convenience wrapper around
481 :meth:`~wuttjamaican.app.AppHandler.render_mako_template()`.
483 :param template: :class:`~mako:mako.template.Template`
484 instance, or name of one to fetch via lookup.
486 This method allows specifying the template by name, in which
487 case the real template object is fetched via lookup.
489 Other args etc. are the same as for the wrapped app handler
490 method.
491 """
492 if isinstance(template, str):
493 template = self.templates.get_template(template)
495 return self.app.render_mako_template(template, context, output_path=output_path)
497 def make_config_file(self, template, output_path, **kwargs):
498 """
499 Write a new config file to the given path, using the given
500 template and context.
502 :param template: :class:`~mako:mako.template.Template`
503 instance, or name of one to fetch via lookup.
505 :param output_path: Path to which output should be written.
507 :param \\**kwargs: Extra context for the template.
509 Some context will be provided automatically for the template,
510 but these may be overridden via the ``**kwargs``:
512 * ``app_title`` - value from
513 :meth:`~wuttjamaican.app.AppHandler.get_title()`.
514 * ``appdir`` - value from
515 :meth:`~wuttjamaican.app.AppHandler.get_appdir()`.
516 * ``db_url`` - poser/dummy value
517 * ``os`` - reference to :mod:`os` module
519 This method is mostly about sorting out the context dict.
520 Once it does that it calls :meth:`render_mako_template()`.
521 """
522 context = {
523 "app_title": self.app.get_title(),
524 "appdir": self.app.get_appdir(),
525 "db_url": "postresql://user:pass@localhost/poser",
526 "os": os,
527 }
528 context.update(kwargs)
529 self.render_mako_template(template, context, output_path=output_path)
530 return output_path
532 def install_db_schema(self, db_url, appdir=None):
533 """
534 First prompt the user; if they agree then apply all Alembic
535 migrations to the database.
537 This method is normally called by :meth:`do_install_steps()`.
538 The end result should be a complete schema, ready for the app
539 to use.
541 :param db_url: :class:`sqlalchemy:sqlalchemy.engine.URL`
542 instance.
544 :param appdir: Optional override for the :term:`app dir`.
545 Probably only useful in tests; this determines where config
546 files should be found. If not specified,
547 :meth:`~wuttjamaican.app.AppHandler.get_appdir()` is called
548 to get it.
550 :returns: ``True`` if user confirmed and schema was installed;
551 otherwise ``False``.
552 """
553 from alembic.util.messaging import ( # pylint: disable=import-outside-toplevel
554 obfuscate_url_pw,
555 )
557 if not self.prompt_bool("install db schema?", True):
558 return False
560 self.rprint()
562 # install db schema
563 appdir = appdir or self.app.get_appdir()
564 cmd = [
565 os.path.join(sys.prefix, "bin", "alembic"),
566 "-c",
567 os.path.join(appdir, "wutta.conf"),
568 "upgrade",
569 "heads",
570 ]
571 subprocess.check_call(cmd)
573 self.rprint(
574 "\n\tdb schema installed to: "
575 f"[bold green]{obfuscate_url_pw(db_url)}[/bold green]"
576 )
577 return True
579 def show_goodbye(self):
580 """
581 Show the final message; this assumes setup completed okay.
583 This is normally called by :meth:`run()`.
584 """
585 self.rprint("\n\t[bold green]initial setup is complete![/bold green]")
587 if self.schema_installed:
588 self.rprint("\n\tyou can run the web app with:")
589 self.rprint(f"\n\t[blue]cd {sys.prefix}[/blue]")
590 self.rprint("\t[blue]bin/wutta -c app/web.conf webapp -r[/blue]")
592 self.rprint()
594 ##############################
595 # console utility functions
596 ##############################
598 def require_prompt_toolkit(self, answer=None): # pylint: disable=empty-docstring
599 """ """
600 try:
601 import prompt_toolkit # pylint: disable=unused-import,import-outside-toplevel
602 except ImportError:
603 value = answer or input(
604 "\nprompt_toolkit is not installed. shall i install it? [Yn] "
605 )
606 value = value.strip()
607 if value and not self.config.parse_bool(value):
608 sys.stderr.write("prompt_toolkit is required; aborting\n")
609 sys.exit(1)
611 subprocess.check_call(
612 [sys.executable, "-m", "pip", "install", "prompt_toolkit"]
613 )
615 # nb. this should now succeed
616 import prompt_toolkit # pylint: disable=import-outside-toplevel
618 def rprint(self, *args, **kwargs):
619 """
620 Convenience wrapper for :func:`rich:rich.print()`.
621 """
622 rich.print(*args, **kwargs)
624 def get_prompt_style(self): # pylint: disable=empty-docstring
625 """ """
626 from prompt_toolkit.styles import ( # pylint: disable=import-outside-toplevel
627 Style,
628 )
630 # message formatting styles
631 return Style.from_dict(
632 {
633 "": "",
634 "bold": "bold",
635 }
636 )
638 def prompt_generic( # pylint: disable=too-many-arguments,too-many-positional-arguments
639 self,
640 info,
641 default=None,
642 is_password=False,
643 is_bool=False,
644 required=False,
645 ):
646 """
647 Prompt the user to get their input.
649 See also :meth:`prompt_bool()`.
651 :param info: String to display (in bold) as prompt text.
653 :param default: Default value to assume if user just presses
654 Enter without providing a value.
656 :param is_bool: Whether the prompt is for a boolean (Y/N)
657 value, vs. a normal text value.
659 :param is_password: Whether the prompt is for a "password" or
660 other sensitive text value. (User input will be masked.)
662 :param required: Whether the value is required (user must
663 provide a value before continuing).
665 :returns: String value provided by the user (or the default),
666 unless ``is_bool`` was requested in which case ``True`` or
667 ``False``.
668 """
669 from prompt_toolkit import prompt # pylint: disable=import-outside-toplevel
671 # build prompt message
672 message = [
673 ("", "\n"),
674 ("class:bold", info),
675 ]
676 if default is not None:
677 if is_bool:
678 message.append(("", f' [{"Y" if default else "N"}]: '))
679 else:
680 message.append(("", f" [{default}]: "))
681 else:
682 message.append(("", ": "))
684 # prompt user for input
685 style = self.get_prompt_style()
686 try:
687 text = prompt(message, style=style, is_password=is_password)
688 except (KeyboardInterrupt, EOFError):
689 self.rprint(
690 "\n\t[bold yellow]operation canceled by user[/bold yellow]\n",
691 file=sys.stderr,
692 )
693 sys.exit(1)
695 if is_bool:
696 if text == "":
697 return default
698 if text.upper() == "Y":
699 return True
700 if text.upper() == "N":
701 return False
702 self.rprint("\n\t[bold yellow]ambiguous, please try again[/bold yellow]\n")
703 return self.prompt_generic(info, default, is_bool=True)
705 if required and not text and not default:
706 return self.prompt_generic(
707 info, default, is_password=is_password, required=True
708 )
710 return text or default
712 def prompt_bool(self, info, default=None):
713 """
714 Prompt the user for a boolean (Y/N) value.
716 Convenience wrapper around :meth:`prompt_generic()` with
717 ``is_bool=True``..
719 :returns: ``True`` or ``False``.
720 """
721 return self.prompt_generic(info, is_bool=True, default=default)