Coverage for .tox/coverage/lib/python3.11/site-packages/wuttjamaican/install.py: 100%
207 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-01-04 23:19 -0600
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 See also
44 :meth:`~wuttjamaican.app.AppHandler.get_install_handler()`.
46 The installer runs interactively via command line, prompting the
47 user for various config settings etc.
49 If installation completes okay the exit code is 0, but if not:
51 * exit code 1 indicates user canceled
52 * exit code 2 indicates sanity check failed
53 * other codes possible if errors occur
55 Usually an app will define e.g. ``poser install`` command which
56 would invoke the install handler's :meth:`run()` method::
58 app = config.get_app()
59 install = app.get_install_handler(pkg_name='poser')
60 install.run()
62 Note that these first 4 attributes may be specified via
63 constructor kwargs:
65 .. attribute:: pkg_name
67 Python package name for the app, e.g. ``poser``.
69 .. attribute:: app_title
71 Display title for the app, e.g. "Poser".
73 .. attribute:: pypi_name
75 Package distribution name, e.g. for PyPI. If not specified one
76 will be guessed.
78 .. attribute:: egg_name
80 Egg name for the app. If not specified one will be guessed.
82 """
84 pkg_name = "poser"
85 app_title = None
86 pypi_name = None
87 egg_name = None
88 schema_installed = False
90 # nb. we prompt the user for this, unless attr already has value
91 wants_continuum = None
93 template_paths = ["wuttjamaican:templates/install"]
95 def __init__(self, config, **kwargs):
96 super().__init__(config)
98 # nb. caller may specify pkg_name etc.
99 self.__dict__.update(kwargs)
101 # some package names we can generate by default
102 if not self.app_title:
103 self.app_title = self.pkg_name
104 if not self.pypi_name:
105 self.pypi_name = self.app_title
106 if not self.egg_name:
107 self.egg_name = self.pypi_name.replace("-", "_")
109 paths = [self.app.resource_path(p) for p in self.template_paths]
111 try:
112 paths.insert(
113 0, self.app.resource_path(f"{self.pkg_name}:templates/install")
114 )
115 except (TypeError, ModuleNotFoundError):
116 pass
118 self.templates = TemplateLookup(directories=paths)
120 def run(self):
121 """
122 Run the interactive command-line installer.
124 This does the following:
126 * check for ``prompt_toolkit`` and maybe ask to install it
127 * call :meth:`show_welcome()`
128 * call :meth:`sanity_check()`
129 * call :meth:`do_install_steps()`
130 * call :meth:`show_goodbye()`
132 Although if a problem is encountered then not all calls may
133 happen.
134 """
135 self.require_prompt_toolkit()
136 self.show_welcome()
137 self.sanity_check()
138 self.schema_installed = False
139 self.do_install_steps()
140 self.show_goodbye()
142 def show_welcome(self):
143 """
144 Show the intro/welcome message, and prompt user to begin the
145 install.
147 This is normally called by :meth:`run()`.
148 """
149 self.rprint(f"\n\t[blue]Welcome to {self.app.get_title()}![/blue]")
150 self.rprint("\n\tThis tool will install and configure the app.")
151 self.rprint(
152 "\n\t[bold italic]NB. You should already have created "
153 "the database in PostgreSQL or MySQL.[/bold italic]"
154 )
156 # shall we continue?
157 if not self.prompt_bool("continue?", True):
158 self.rprint()
159 sys.exit(1)
161 def sanity_check(self):
162 """
163 Perform various sanity checks before doing the install. If
164 any problem is found the installer should exit with code 2.
166 This is normally called by :meth:`run()`.
168 The default logic here just calls :meth:`check_appdir()`.
169 """
170 self.check_appdir()
172 def check_appdir(self):
173 """
174 Check if the :term:`app dir` already exists; exit with code 2
175 if so.
177 This is normally called from :meth:`sanity_check()`.
178 """
179 # appdir must not yet exist
180 appdir = os.path.join(sys.prefix, "app")
181 if os.path.exists(appdir):
182 self.rprint(f"\n\t[bold red]appdir already exists:[/bold red] {appdir}\n")
183 sys.exit(2)
185 def do_install_steps(self):
186 """
187 Perform the real installation steps.
189 This method is called by :meth:`run()` and does the following:
191 * call :meth:`prompt_user_for_context()` to collect DB info etc.
192 * call :meth:`make_template_context()` to use when generating output
193 * call :meth:`make_appdir()` to create app dir with config files
194 * call :meth:`install_db_schema()` to (optionally) create tables in DB
195 """
196 # prompt user / get context
197 context = self.prompt_user_for_context()
198 context = self.make_template_context(**context)
200 # make the appdir
201 self.make_appdir(context)
203 # install db schema if user likes
204 self.schema_installed = self.install_db_schema(context["db_url"])
206 def prompt_user_for_context(self):
207 """
208 This is responsible for initial user prompts.
210 This happens early in the install, so this method can verify
211 the info, e.g. test the DB connection, but should not write
212 any files as the app dir may not exist yet.
214 Default logic calls :meth:`get_db_url()` for the DB
215 connection, then may ask about Wutta-Continuum data
216 versioning. (The latter is skipped if the package is
217 missing.)
219 Subclass should override this method if they need different
220 prompting logic. The return value should always include at
221 least these 2 items:
223 * ``db_url`` - URL for the DB connection
224 * ``wants_continuum`` - whether data versioning should be enabled
226 :returns: Dict of template context
227 """
228 # db info
229 db_url = self.get_db_url()
231 # continuum
232 if self.wants_continuum is None:
233 try:
234 import wutta_continuum # pylint: disable=import-outside-toplevel,unused-import
235 except ImportError:
236 self.wants_continuum = False
237 else:
238 self.wants_continuum = self.prompt_bool(
239 "use continuum for data versioning?", default=False
240 )
242 return {"db_url": db_url, "wants_continuum": self.wants_continuum}
244 def get_db_url(self):
245 """
246 This must return the DB engine URL.
248 Default logic will prompt the user for hostname, port, DB name
249 and credentials. It then assembles the URL from those parts.
251 This method will also test the DB connection. If it fails,
252 the install is aborted.
254 This method is normally called by
255 :meth:`prompt_user_for_context()`.
257 :returns: SQLAlchemy engine URL (as object or string)
258 """
259 # get db info/url
260 dbinfo = self.get_dbinfo()
261 db_url = dbinfo.get("db_url")
262 if not db_url:
263 db_url = self.make_db_url(dbinfo)
265 # test db connection
266 self.rprint("\n\ttesting db connection... ", end="")
267 error = self.test_db_connection(db_url)
268 if error:
269 self.rprint("[bold red]cannot connect![/bold red] ..error was:")
270 self.rprint(f"\n{error}")
271 self.rprint("\n\t[bold yellow]aborting mission[/bold yellow]\n")
272 sys.exit(1)
273 self.rprint("[bold green]good[/bold green]")
275 return db_url
277 def get_dbinfo(self): # pylint: disable=missing-function-docstring
278 dbinfo = {}
280 # main info
281 dbinfo["dbtype"] = self.prompt_generic("db type", "postgresql")
282 dbinfo["dbhost"] = self.prompt_generic("db host", "localhost")
283 default_port = "3306" if dbinfo["dbtype"] == "mysql" else "5432"
284 dbinfo["dbport"] = self.prompt_generic("db port", default_port)
285 dbinfo["dbname"] = self.prompt_generic("db name", self.pkg_name)
286 dbinfo["dbuser"] = self.prompt_generic("db user", self.pkg_name)
288 # password
289 dbinfo["dbpass"] = None
290 while not dbinfo["dbpass"]:
291 dbinfo["dbpass"] = self.prompt_generic("db pass", is_password=True)
293 return dbinfo
295 def make_db_url(self, dbinfo): # pylint: disable=empty-docstring
296 """ """
297 from sqlalchemy.engine import URL # pylint: disable=import-outside-toplevel
299 if dbinfo["dbtype"] == "mysql":
300 drivername = "mysql+mysqlconnector"
301 else:
302 drivername = "postgresql+psycopg2"
304 return URL.create(
305 drivername=drivername,
306 username=dbinfo["dbuser"],
307 password=dbinfo["dbpass"],
308 host=dbinfo["dbhost"],
309 port=dbinfo["dbport"],
310 database=dbinfo["dbname"],
311 )
313 def test_db_connection(self, url): # pylint: disable=empty-docstring
314 """ """
315 import sqlalchemy as sa # pylint: disable=import-outside-toplevel
317 engine = sa.create_engine(url)
319 # check for random table; does not matter if it exists, we
320 # just need to test interaction and this is a neutral way
321 try:
322 sa.inspect(engine).has_table("whatever")
323 except Exception as error: # pylint: disable=broad-exception-caught
324 return str(error)
325 return None
327 def make_template_context(self, **kwargs):
328 """
329 This must return a dict to be used as global template context
330 when generating output (e.g. config) files.
332 This method is normally called by :meth:`do_install_steps()`.
333 The ``context`` returned is then passed to
334 :meth:`render_mako_template()`.
336 Note these first 2 params are not explicitly listed in the
337 method signature; they are required nonetheless.
339 :param db_url: This must be a string URL for the DB engine.
341 :param wants_continuum: Whether data versioning should be
342 enabled within the config.
344 :param \\**kwargs: Extra template context.
346 :returns: Dict for global template context.
348 The final context dict should include at least:
350 * ``envdir`` - value from :data:`python:sys.prefix`
351 * ``envname`` - "last" dirname from ``sys.prefix``
352 * ``pkg_name`` - value from :attr:`pkg_name`
353 * ``app_title`` - value from :attr:`app_title`
354 * ``pypi_name`` - value from :attr:`pypi_name`
355 * ``egg_name`` - value from :attr:`egg_name`
356 * ``appdir`` - ``app`` folder under ``sys.prefix``
357 * ``db_url`` - value from ``kwargs``
358 * ``wants_continuum`` - value from ``kwargs``
359 """
360 envname = os.path.basename(sys.prefix)
361 appdir = os.path.join(sys.prefix, "app")
363 db_url = kwargs.pop("db_url")
364 if not isinstance(db_url, str):
365 db_url = db_url.render_as_string(hide_password=False)
367 context = {
368 "envdir": sys.prefix,
369 "envname": envname,
370 "pkg_name": self.pkg_name,
371 "app_title": self.app_title,
372 "pypi_name": self.pypi_name,
373 "appdir": appdir,
374 "egg_name": self.egg_name,
375 "db_url": db_url,
376 }
377 context.update(kwargs)
378 return context
380 def make_appdir(self, context, appdir=None):
381 """
382 Create the app folder structure and generate config files.
384 This method is normally called by :meth:`do_install_steps()`.
386 :param context: Template context dict, i.e. from
387 :meth:`make_template_context()`.
389 The default logic will create a structure as follows, assuming
390 ``/venv`` is the path to the virtual environment:
392 .. code-block:: none
394 /venv/
395 └── app/
396 ├── cache/
397 ├── data/
398 ├── log/
399 ├── work/
400 ├── wutta.conf
401 ├── web.conf
402 └── upgrade.sh
404 File templates for this come from
405 ``wuttjamaican:templates/install`` by default.
407 This method calls
408 :meth:`~wuttjamaican.app.AppHandler.make_appdir()` for the
409 basic structure and then :meth:`write_all_config_files()` for
410 the gory details.
411 """
412 # app handler makes appdir proper
413 appdir = appdir or self.app.get_appdir()
414 self.app.make_appdir(appdir)
416 # but then we also generate some files...
417 self.write_all_config_files(appdir, context)
419 self.rprint(f"\n\tappdir created at: [bold green]{appdir}[/bold green]")
421 def write_all_config_files(self, appdir, context):
422 """
423 This method should write all config files within the app dir.
424 It's called from :meth:`make_appdir()`.
426 Subclass can override this for specialized installers.
428 Note that the app dir may or may not be newly-created, when
429 this method is called. Some installers may support a
430 "refresh" of the existing app dir.
432 Default logic (over)writes 3 files:
434 * ``wutta.conf``
435 * ``web.cof``
436 * ``upgrade.sh``
437 """
438 self.write_wutta_conf(appdir, context)
439 self.write_web_conf(appdir, context)
440 self.write_upgrade_sh(appdir, context)
442 def write_wutta_conf(
443 self, appdir, context
444 ): # pylint: disable=missing-function-docstring
445 self.make_config_file(
446 "wutta.conf.mako", os.path.join(appdir, "wutta.conf"), **context
447 )
449 def write_web_conf(
450 self, appdir, context
451 ): # pylint: disable=missing-function-docstring
452 web_context = dict(context)
453 web_context.setdefault("beaker_key", context.get("pkg_name", "poser"))
454 web_context.setdefault("beaker_secret", "TODO_YOU_SHOULD_CHANGE_THIS")
455 web_context.setdefault("pyramid_host", "0.0.0.0")
456 web_context.setdefault("pyramid_port", "9080")
457 self.make_config_file(
458 "web.conf.mako", os.path.join(appdir, "web.conf"), **web_context
459 )
461 def write_upgrade_sh(
462 self, appdir, context
463 ): # pylint: disable=missing-function-docstring
464 template = self.templates.get_template("upgrade.sh.mako")
465 output_path = os.path.join(appdir, "upgrade.sh")
466 self.render_mako_template(template, context, output_path=output_path)
467 os.chmod(
468 output_path,
469 stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH,
470 )
472 def render_mako_template(
473 self,
474 template,
475 context,
476 output_path=None,
477 ):
478 """
479 Convenience wrapper around
480 :meth:`~wuttjamaican.app.AppHandler.render_mako_template()`.
482 :param template: :class:`~mako:mako.template.Template`
483 instance, or name of one to fetch via lookup.
485 This method allows specifying the template by name, in which
486 case the real template object is fetched via lookup.
488 Other args etc. are the same as for the wrapped app handler
489 method.
490 """
491 if isinstance(template, str):
492 template = self.templates.get_template(template)
494 return self.app.render_mako_template(template, context, output_path=output_path)
496 def make_config_file(self, template, output_path, **kwargs):
497 """
498 Write a new config file to the given path, using the given
499 template and context.
501 :param template: :class:`~mako:mako.template.Template`
502 instance, or name of one to fetch via lookup.
504 :param output_path: Path to which output should be written.
506 :param \\**kwargs: Extra context for the template.
508 Some context will be provided automatically for the template,
509 but these may be overridden via the ``**kwargs``:
511 * ``app_title`` - value from
512 :meth:`~wuttjamaican.app.AppHandler.get_title()`.
513 * ``appdir`` - value from
514 :meth:`~wuttjamaican.app.AppHandler.get_appdir()`.
515 * ``db_url`` - poser/dummy value
516 * ``os`` - reference to :mod:`os` module
518 This method is mostly about sorting out the context dict.
519 Once it does that it calls :meth:`render_mako_template()`.
520 """
521 context = {
522 "app_title": self.app.get_title(),
523 "appdir": self.app.get_appdir(),
524 "db_url": "postresql://user:pass@localhost/poser",
525 "os": os,
526 }
527 context.update(kwargs)
528 self.render_mako_template(template, context, output_path=output_path)
529 return output_path
531 def install_db_schema(self, db_url, appdir=None):
532 """
533 First prompt the user, but if they agree then apply all
534 Alembic migrations to the configured database.
536 This method is normally called by :meth:`do_install_steps()`.
537 The end result should be a complete schema, ready for the app
538 to use.
540 :param db_url: :class:`sqlalchemy:sqlalchemy.engine.URL`
541 instance.
542 """
543 from alembic.util.messaging import ( # pylint: disable=import-outside-toplevel
544 obfuscate_url_pw,
545 )
547 if not self.prompt_bool("install db schema?", True):
548 return False
550 self.rprint()
552 # install db schema
553 appdir = appdir or self.app.get_appdir()
554 cmd = [
555 os.path.join(sys.prefix, "bin", "alembic"),
556 "-c",
557 os.path.join(appdir, "wutta.conf"),
558 "upgrade",
559 "heads",
560 ]
561 subprocess.check_call(cmd)
563 self.rprint(
564 "\n\tdb schema installed to: "
565 f"[bold green]{obfuscate_url_pw(db_url)}[/bold green]"
566 )
567 return True
569 def show_goodbye(self):
570 """
571 Show the final message; this assumes setup completed okay.
573 This is normally called by :meth:`run()`.
574 """
575 self.rprint("\n\t[bold green]initial setup is complete![/bold green]")
577 if self.schema_installed:
578 self.rprint("\n\tyou can run the web app with:")
579 self.rprint(f"\n\t[blue]cd {sys.prefix}[/blue]")
580 self.rprint("\t[blue]bin/wutta -c app/web.conf webapp -r[/blue]")
582 self.rprint()
584 ##############################
585 # console utility functions
586 ##############################
588 def require_prompt_toolkit(self, answer=None): # pylint: disable=empty-docstring
589 """ """
590 try:
591 import prompt_toolkit # pylint: disable=unused-import,import-outside-toplevel
592 except ImportError:
593 value = answer or input(
594 "\nprompt_toolkit is not installed. shall i install it? [Yn] "
595 )
596 value = value.strip()
597 if value and not self.config.parse_bool(value):
598 sys.stderr.write("prompt_toolkit is required; aborting\n")
599 sys.exit(1)
601 subprocess.check_call(
602 [sys.executable, "-m", "pip", "install", "prompt_toolkit"]
603 )
605 # nb. this should now succeed
606 import prompt_toolkit # pylint: disable=import-outside-toplevel
608 def rprint(self, *args, **kwargs):
609 """
610 Convenience wrapper for :func:`rich:rich.print()`.
611 """
612 rich.print(*args, **kwargs)
614 def get_prompt_style(self): # pylint: disable=empty-docstring
615 """ """
616 from prompt_toolkit.styles import ( # pylint: disable=import-outside-toplevel
617 Style,
618 )
620 # message formatting styles
621 return Style.from_dict(
622 {
623 "": "",
624 "bold": "bold",
625 }
626 )
628 def prompt_generic( # pylint: disable=too-many-arguments,too-many-positional-arguments
629 self,
630 info,
631 default=None,
632 is_password=False,
633 is_bool=False,
634 required=False,
635 ):
636 """
637 Prompt the user to get their input.
639 See also :meth:`prompt_bool()`.
641 :param info: String to display (in bold) as prompt text.
643 :param default: Default value to assume if user just presses
644 Enter without providing a value.
646 :param is_bool: Whether the prompt is for a boolean (Y/N)
647 value, vs. a normal text value.
649 :param is_password: Whether the prompt is for a "password" or
650 other sensitive text value. (User input will be masked.)
652 :param required: Whether the value is required (user must
653 provide a value before continuing).
655 :returns: String value provided by the user (or the default),
656 unless ``is_bool`` was requested in which case ``True`` or
657 ``False``.
658 """
659 from prompt_toolkit import prompt # pylint: disable=import-outside-toplevel
661 # build prompt message
662 message = [
663 ("", "\n"),
664 ("class:bold", info),
665 ]
666 if default is not None:
667 if is_bool:
668 message.append(("", f' [{"Y" if default else "N"}]: '))
669 else:
670 message.append(("", f" [{default}]: "))
671 else:
672 message.append(("", ": "))
674 # prompt user for input
675 style = self.get_prompt_style()
676 try:
677 text = prompt(message, style=style, is_password=is_password)
678 except (KeyboardInterrupt, EOFError):
679 self.rprint(
680 "\n\t[bold yellow]operation canceled by user[/bold yellow]\n",
681 file=sys.stderr,
682 )
683 sys.exit(1)
685 if is_bool:
686 if text == "":
687 return default
688 if text.upper() == "Y":
689 return True
690 if text.upper() == "N":
691 return False
692 self.rprint("\n\t[bold yellow]ambiguous, please try again[/bold yellow]\n")
693 return self.prompt_generic(info, default, is_bool=True)
695 if required and not text and not default:
696 return self.prompt_generic(
697 info, default, is_password=is_password, required=True
698 )
700 return text or default
702 def prompt_bool(self, info, default=None):
703 """
704 Prompt the user for a boolean (Y/N) value.
706 Convenience wrapper around :meth:`prompt_generic()` with
707 ``is_bool=True``..
709 :returns: ``True`` or ``False``.
710 """
711 return self.prompt_generic(info, is_bool=True, default=default)