gramex

Parses command line / configuration and runs Gramex.

Running gramex on the command line calls:

  1. gramex.commandline to parse command line arguments
  2. gramex.init to parse the configuration and start Gramex

This module also has

commandline(args=None)

Run Gramex from the command line.

Parameters:

Name Type Description Default
args List[str]

Command line arguments. If not provided, uses sys.argv

None

Gramex can be run in 2 ways, both of which call gramex.commandline:

  1. python -m gramex, which runs __main__.py.
  2. gramex, which runs console_scripts in setup.py

gramex -V and gramex --version exit after printing the Gramex version.

The positional arguments call different functions:

Keyword arguments are passed to the function, e.g. gramex service install --startup=auto calls gramex.install.service('install', startup='auto').

Keyword arguments with ‘.’ are split into sub-keys, e.g. gramex --listen.port 80 becomes init(listen={"port": 80}).

Values are parsed as YAML, e.g. null becomes None.

If the keyword arguments include --help, it prints the usage of that function and exits.

Source code in gramex\__init__.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def commandline(args: List[str] = None):
    '''Run Gramex from the command line.

    Parameters:
        args: Command line arguments. If not provided, uses `sys.argv`

    Gramex can be run in 2 ways, both of which call [gramex.commandline][]:

    1. `python -m gramex`, which runs `__main__.py`.
    2. `gramex`, which runs `console_scripts` in `setup.py`

    `gramex -V` and `gramex --version` exit after printing the Gramex version.

    The positional arguments call different functions:

    - `gramex` runs Gramex and calls [gramex.init][]
    - `gramex init` creates a new Gramex project and calls [gramex.install.init][]
    - `gramex service` creates a Windows service and calls [gramex.install.service][]
    - ... etc. (Run `gramex help` for more)

    Keyword arguments are passed to the function, e.g.
    `gramex service install --startup=auto` calls
    `gramex.install.service('install', startup='auto')`.

    Keyword arguments with '.' are split into sub-keys, e.g.
    `gramex --listen.port 80` becomes `init(listen={"port": 80})`.

    Values are parsed as YAML, e.g. `null` becomes `None`.

    If the keyword arguments include `--help`, it prints the usage of that function and exits.
    '''
    commands = sys.argv[1:] if args is None else args

    # First, setup log: service at INFO to log progress. App's log: may override this later.
    log_config = (+PathConfig(paths['source'] / 'gramex.yaml')).get('log', AttrDict())
    log_config.loggers.gramex.level = logging.INFO
    from . import services

    services.log(log_config)

    # kwargs has all optional command line args as a dict of values / lists.
    # args has all positional arguments as a list.
    kwargs = parse_command_line(commands)
    args = kwargs.pop('_')

    # If -V or --version is specified, print a message and end
    if 'V' in kwargs or 'version' in kwargs:
        pyver = '{0}.{1}.{2}'.format(*sys.version_info[:3])
        msg = [
            f'Gramex version: {__version__}',
            f'Gramex path: {paths["source"]}',
            f'Python version: {pyver}',
            f'Python path: {sys.executable}',
        ]
        return console(msg='\n'.join(msg))

    # Any positional argument is treated as a gramex command
    if len(args) > 0:
        base_command = args.pop(0).lower()
        method = 'install' if base_command == 'update' else base_command
        if method in {
            'install',
            'uninstall',
            'setup',
            'run',
            'service',
            'init',
            'mail',
            'license',
            'features',
            'complexity',
        }:
            import gramex.install

            if 'help' in kwargs:
                return console(msg=gramex.install.show_usage(method))
            return getattr(gramex.install, method)(args=args, kwargs=kwargs)
        raise NotImplementedError(f'Unknown gramex command: {base_command}')
    elif 'help' in kwargs:
        return console(msg=help.strip().format(**globals()))

    # Use current dir as base (where gramex is run from) if there's a gramex.yaml.
    if not os.path.isfile('gramex.yaml'):
        return console(msg='No gramex.yaml. See https://gramener.com/gramex/guide/')

    pyver = sys.version.replace('\n', ' ')
    app_log.info(f'Gramex {__version__} | {os.getcwd()} | Python {pyver}')

    # Run gramex.init(cmd={command line arguments like YAML variables})
    # --log.* settings are moved to log.loggers.gramex.*
    #   E.g. --log.level => log.loggers.gramex.level
    # --* remaining settings are moved to app.*
    #   E.g. --watch => app.watch
    config = AttrDict(app=kwargs)
    if kwargs.get('log'):
        config.log = AttrDict(loggers=AttrDict(gramex=kwargs.pop('log')))
        if 'level' in config.log.loggers.gramex:
            config.log.loggers.gramex.level = config.log.loggers.gramex.level.upper()
    return init(cmd=config)

init(force_reload=False, kwargs)

Load Gramex configurations and start / restart the Gramex instance.

Parameters:

Name Type Description Default
force_reload bool

Reload services even config hasn’t changed

False
**kwargs dict

Overrides config

{}

gramex.init() loads configurations from 3 sources, which override each other:

  1. Gramex’s gramex.yaml
  2. Current directory’s gramex.yaml
  3. The kwargs

Then it calls each gramex.services with its configuration.

It can be called multiple times. For efficiency, a services function is called only if its configuration has changed or force_reload=True.

If a kwarg value is:

  • a string, it’s loaded as-is
  • a Path pointing to a file, it’s loaded as a YAML config file
  • a Path pointing to a directory, it loads a gramex.yaml file from that directory
Source code in gramex\__init__.py
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def init(force_reload: bool = False, **kwargs) -> None:
    '''Load Gramex configurations and start / restart the Gramex instance.

    Parameters:
        force_reload (bool): Reload services even config hasn't changed
        **kwargs (dict): Overrides config

    `gramex.init()` loads configurations from 3 sources, which override each other:

    1. Gramex's `gramex.yaml`
    2. Current directory's `gramex.yaml`
    3. The `kwargs`

    Then it calls each [gramex.services][] with its configuration.

    It can be called multiple times. For efficiency, a services function is called only if its
    configuration has changed or `force_reload=True`.

    If a kwarg value is:

    - a string, it's loaded as-is
    - a Path pointing to a file, it's loaded as a YAML config file
    - a Path pointing to a directory, it loads a `gramex.yaml` file from that directory
    '''
    # Set up secrets from .secrets.yaml, if any
    try:
        setup_secrets(paths['base'] / '.secrets.yaml')
    except Exception as e:
        app_log.exception(e)

    # Add base path locations where config files are found to sys.path.
    # This allows variables: to import files from folder where configs are defined.
    sys.path[:] = _sys_path + [
        str(path.absolute()) for path in paths.values() if isinstance(path, Path)
    ]

    # Reset variables
    variables.clear()
    variables.update(setup_variables())

    # Initialise configuration layers with provided configurations
    # AttrDicts are updated as-is. Paths are converted to PathConfig
    paths.update(kwargs)
    for key, val in paths.items():
        if isinstance(val, Path):
            if val.is_dir():
                val = val / 'gramex.yaml'
            val = PathConfig(val)
        config_layers[key] = val

    # Locate all config files
    config_files = set()
    for path_config in config_layers.values():
        if hasattr(path_config, '__info__'):
            for pathinfo in path_config.__info__.imports:
                config_files.add(pathinfo.path)
    config_files = list(config_files)

    # Add config file folders to sys.path
    sys.path[:] = _sys_path + [str(path.absolute().parent) for path in config_files]

    from . import services

    globals()['service'] = services.info  # gramex.service = gramex.services.info

    # Override final configurations
    appconfig.clear()
    appconfig.update(+config_layers)
    # If --settings.debug, override root and Gramex loggers to show debug messages
    if appconfig.app.get('settings', {}).get('debug', False):
        appconfig.log.root.level = appconfig.log.loggers.gramex.level = logging.DEBUG

    # Set up a watch on config files (including imported files)
    if appconfig.app.get('watch', True):
        from services import watcher

        watcher.watch('gramex-reconfig', paths=config_files, on_modified=lambda event: init())

    # Run all valid services. (The "+" before config_chain merges the chain)
    # Services may return callbacks to be run at the end
    for key, val in appconfig.items():
        if key not in conf or conf[key] != val or force_reload:
            if hasattr(services, key):
                app_log.debug(f'Loading service: {key}')
                conf[key] = prune_keys(val, {'comment'})
                callback = getattr(services, key)(conf[key])
                if callable(callback):
                    callbacks[key] = callback
            else:
                app_log.error(f'No service named {key}')

    # Run the callbacks. Specifically, the app service starts the Tornado ioloop
    for key in (+config_layers).keys():
        if key in callbacks:
            app_log.debug(f'Running callback: {key}')
            callbacks[key]()

shutdown()

Shut down the running Gramex instance.

Source code in gramex\__init__.py
322
323
324
325
326
327
328
329
330
def shutdown():
    '''Shut down the running Gramex instance.'''
    from . import services

    ioloop = services.info.main_ioloop
    if ioloop_running(ioloop):
        app_log.info('Shutting down Gramex...')
        # Shut down Gramex in a thread-safe way. add_callback is the ONLY thread-safe method
        ioloop.add_callback(ioloop.stop)

gramex_update(url)

Check if a newer version of Gramex is available. If yes, log a warning.

Parameters:

Name Type Description Default
url str

URL to check for new version

required

Gramex uses https://gramener.com/gramex-update/ as the URL to check for new versions.

Source code in gramex\__init__.py
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def gramex_update(url: str):
    '''Check if a newer version of Gramex is available. If yes, log a warning.

    Parameters:
        url: URL to check for new version

    Gramex uses <https://gramener.com/gramex-update/> as the URL to check for new versions.
    '''
    import time
    import requests
    import platform
    from . import services

    if not services.info.eventlog:
        return app_log.error('eventlog: service is not running. So Gramex update is disabled')

    query = services.info.eventlog.query
    update = query('SELECT * FROM events WHERE event="update" ORDER BY time DESC LIMIT 1')
    delay = 24 * 60 * 60  # Wait for one day before updates
    if update and time.time() < update[0]['time'] + delay:
        return app_log.debug('Gramex update ran recently. Deferring check.')

    meta = {
        'dir': variables.get('GRAMEXDATA'),
        'uname': platform.uname(),
    }
    if update:
        events = query('SELECT * FROM events WHERE time > ? ORDER BY time', (update[0]['time'],))
    else:
        events = query('SELECT * FROM events')
    logs = [dict(log, **meta) for log in events]

    r = requests.post(url, data=json.dumps(logs))
    r.raise_for_status()
    update = r.json()
    server_version = update['version']
    if version.parse(server_version) > version.parse(__version__):
        app_log.error(f'Gramex {server_version} is available. https://gramener.com/gramex/guide/')
    elif version.parse(server_version) < version.parse(__version__):
        app_log.warning(f'Gramex {__version__} is ahead of stable {server_version}')
    else:
        app_log.debug(f'Gramex {__version__} is up to date')
    services.info.eventlog.add('update', update)
    return {'logs': logs, 'response': update}

log(args, kwargs)

Logs structured information for future reference.

Examples:

>>> gramex.log(level='INFO', x=1, msg='abc')

This logs {level: INFO, x: 1, msg: abc} into a logging queue.

If a gramexlog service like ElasticSearch has been configured, it will periodically flush the logs into the server.

Source code in gramex\__init__.py
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
def log(*args, **kwargs):
    '''Logs structured information for future reference.

    Examples:
        >>> gramex.log(level='INFO', x=1, msg='abc')

        This logs `{level: INFO, x: 1, msg: abc}` into a logging queue.

        If a `gramexlog` service like ElasticSearch has been configured, it will periodically flush
        the logs into the server.
    '''
    from . import services

    # gramexlog() positional arguments may have a handler and app (in any order)
    # The app defaults to the first gramexlog:
    handler, app = None, services.info.gramexlog.get('defaultapp', None)
    for arg in args:
        # Pretend that anything that has a .args is a handler
        if hasattr(getattr(arg, 'args', None), 'items'):
            handler = arg
        # ... and anything that's a string is an index name. The last string overrides all
        elif isinstance(arg, str):
            app = arg
    # If the user logs into an unknown app, stop immediately
    try:
        conf = services.info.gramexlog.apps[app]
    except KeyError:
        raise ValueError(f'gramexlog: no config for {app}')

    # Add all URL query parameters. In case of multiple values, capture the last
    if handler:
        kwargs.update({key: val[-1] for key, val in handler.args.items()})
    # Add additional keys specified in gramex.yaml via keys:
    kwargs.update(conf.extra_keys(handler))
    conf.queue.append(kwargs)

console(msg)

Write message to console. An alias for print().

Source code in gramex\__init__.py
416
417
418
def console(msg):
    '''Write message to console. An alias for print().'''
    print(msg)  # noqa