~sirn/fanboi2

ref: 511ea397b10507eadb4f9e4b8180984ad7d269ae fanboi2/fanboi2/cmd/ctl.py -rw-r--r-- 2.9 KiB
511ea397 — Kridsada Thanabulpong chore: relicense to AGPLv3 1 year, 8 days ago 
                                                                                        
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
import argparse
import code
import multiprocessing
import sys

from ..version import __VERSION__, __PYRAMID__


SHELL_BANNER = """\
      ___  ___
    / /     / /
   /_/_   _/_/    fanboi2 %s
  / /   / /       pyramid %s
 /_/   /_/__

Loaded locals:

    app           Fanboi2 WSGI application
    config        Pyramid configurator
    registry      Pyramid registry
    request       Pyramid request context
    root          Pyramid root
    root_factory  Pyramid root factory
"""


def run_shell(args):
    """Run the interactive shell for the application."""
    from ..wsgi import app, config
    import pyramid.scripting

    with pyramid.scripting.prepare() as env:
        code.interact(
            SHELL_BANNER % (__VERSION__, __PYRAMID__),
            local={"app": app, "config": config, **env},
        )


def run_serve(args):
    """Run the web server for the application."""
    from gunicorn.app.base import BaseApplication
    from ..wsgi import app as wsgi_app

    class FbserveApplication(BaseApplication):
        def __init__(self, app, options=None):
            if not options:
                options = {}
            self.options = options
            self.application = app
            super(FbserveApplication, self).__init__()

        def load_config(self):
            for key, value in self.options.items():
                if key in self.cfg.settings and value is not None:
                    self.cfg.set(key.lower(), value)

        def load(self):
            return self.application

    options = {
        "bind": "%s:%s" % (args.host, args.port),
        "max_requests": args.max_requests,
        "reload": args.reload,
        "threads": args.threads,
        "workers": args.workers,
    }

    FbserveApplication(wsgi_app, options).run()


def run_gensecret(args):
    """Generates a NaCl secret."""
    from pyramid_nacl_session import generate_secret

    print(generate_secret(as_hex=True).decode("utf-8"))
    sys.exit(0)


def main():
    """Parse the command line arguments."""
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="subparser", title="subcommands")

    shell = subparsers.add_parser("shell")
    shell.set_defaults(func=run_shell)

    cpu_count = multiprocessing.cpu_count()
    serve = subparsers.add_parser("serve")
    serve.add_argument("--port", type=int, default=6543)
    serve.add_argument("--host", default="0.0.0.0")
    serve.add_argument("--workers", type=int, default=(cpu_count * 2) + 1)
    serve.add_argument("--threads", type=int, default=1)
    serve.add_argument("--max-requests", type=int, default=1000)
    serve.add_argument("--reload", action="store_true")
    serve.set_defaults(func=run_serve)

    gensecret = subparsers.add_parser("gensecret")
    gensecret.set_defaults(func=run_gensecret)

    args = parser.parse_args()
    if args.subparser is None:
        parser.print_usage()
        sys.exit(1)

    args.func(args)