#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# Isomer - The distributed application framework
# ==============================================
# Copyright (C) 2011-2020 Heiko 'riot' Weinen <riot@c-base.org> and others.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Module: SchemaManager
=====================
"""
from circuits import Event
from isomer.events.client import send
from isomer.events.schemamanager import get, all, configuration
from isomer.debugger import cli_register_event
from isomer.component import ConfigurableComponent
from isomer.schemastore import schemastore, l10n_schemastore, configschemastore
from isomer.logger import warn, debug, error # , hilight
from isomer.component import handler
[docs]class cli_schema(Event):
"""Display a specified schema"""
pass
[docs]class cli_schemata(Event):
"""Display all registered schemata"""
pass
[docs]class cli_default_perms(Event):
"""Display all schemata default permission roles"""
pass
[docs]class SchemaManager(ConfigurableComponent):
"""
Handles schemata requests from clients.
"""
channel = "isomer-web"
configprops = {}
[docs] def __init__(self, *args):
super(SchemaManager, self).__init__("SM", *args)
self.fireEvent(cli_register_event("schema", cli_schema))
self.fireEvent(cli_register_event("schemata", cli_schemata))
self.fireEvent(cli_register_event("form", cli_form))
self.fireEvent(cli_register_event("forms", cli_forms))
self.fireEvent(cli_register_event("permissions_default", cli_default_perms))
[docs] @handler("cli_schemata")
def cli_schemata_list(self, *args):
"""Display a list of registered schemata"""
self.log(
"Registered schemata languages:", ",".join(sorted(l10n_schemastore.keys()))
)
self.log("Registered Schemata:", ",".join(sorted(schemastore.keys())))
if "-c" in args or "-config" in args:
self.log(
"Registered Configuration Schemata:",
",".join(sorted(configschemastore.keys())),
pretty=True,
)
[docs] @handler("cli_schema")
def cli_schema(self, *args):
"""Display a single schema definition"""
key = None
if len(args) > 1:
key = args[1]
args = list(args)
if "-config" in args or "-c" in args:
store = configschemastore
try:
args.remove("-c")
args.remove("-config")
except ValueError:
pass
else:
store = schemastore
def output(schema_name):
self.log("%s :" % schema_name)
if key == "props":
self.log(store[schema_name]["schema"]["properties"], pretty=True)
elif key == "perms":
try:
self.log(store[schema_name]["schema"]["roles_create"], pretty=True)
except KeyError:
self.log("Schema", schema_name, "has no role for creation", lvl=warn)
try:
self.log(
store[schema_name]["schema"]["properties"]["perms"]["properties"],
pretty=True,
)
except KeyError:
self.log("Schema", schema_name, "has no permissions", lvl=warn)
else:
self.log(store[schema_name]["schema"], pretty=True)
if "*" in args:
for schema in store:
output(schema)
else:
output(args[0])
[docs] @handler("cli_default_perms")
def cli_default_perms(self, *args):
"""Show default permissions for all schemata"""
for key, item in schemastore.items():
# self.log(item, pretty=True)
if item["schema"].get("no_perms", False):
self.log("Schema without permissions:", key)
continue
try:
perms = item["schema"]["properties"]["perms"]["properties"]
if perms == {}:
self.log("Schema:", item, pretty=True)
self.log(
"Schema:",
key,
"read",
perms["read"]["default"],
"write",
perms["write"]["default"],
"list",
perms["list"]["default"],
"create",
item["schema"]["roles_create"],
)
except KeyError as e:
self.log("Fishy schema found:", key, e, lvl=error)
self.log(item, pretty=True)
[docs] @handler("ready")
def ready(self):
"""Sets up the application after startup."""
self.log(
"Got",
len(schemastore),
"data and",
len(configschemastore),
"component schemata.",
lvl=debug,
)
# pprint(schemastore.keys())
# pprint(configschemastore.keys())
[docs] @handler(all)
def all(self, event):
"""Return all known schemata to the requesting client"""
self.log("Schemarequest for all schemata from", event.user, lvl=debug)
response = {
"component": "isomer.events.schemamanager",
"action": "all",
"data": l10n_schemastore[event.client.language],
}
self.fireEvent(send(event.client.uuid, response))
[docs] @handler(get)
def get(self, event):
"""Return a single schema"""
self.log("Schemarequest for", event.data, "from", event.user, lvl=debug)
if event.data in schemastore:
response = {
"component": "isomer.events.schemamanager",
"action": "get",
"data": l10n_schemastore[event.client.language][event.data],
}
self.fireEvent(send(event.client.uuid, response))
else:
self.log("Unavailable schema requested!", lvl=warn)
[docs] @handler(configuration)
def configuration(self, event):
"""Return all configurable components' schemata"""
try:
self.log(
"Schemarequest for all configuration schemata from",
event.user.account.name,
lvl=debug,
)
response = {
"component": "isomer.events.schemamanager",
"action": "configuration",
"data": configschemastore,
}
self.fireEvent(send(event.client.uuid, response))
except Exception as e:
self.log("ERROR:", e)