blob: 96e8dc21e0e4541af7247b8745dd028c9a361481 [file] [log] [blame]
# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import print_function
import pdb
import re
from inflect import engine as inflect_engine_class
inflect_engine = inflect_engine_class()
class FieldNotFound(Exception):
def __init__(self, message):
super(FieldNotFound, self).__init__(message)
def xproto_debug(**kwargs):
print(kwargs)
pdb.set_trace()
def xproto_unquote(s):
return unquote(s)
def unquote(s):
if s.startswith('"') and s.endswith('"'):
return s[1:-1]
else:
return s
def xproto_singularize(field):
try:
# The user has set a singular, as an exception that cannot be handled automatically
singular = field["options"]["singular"]
singular = unquote(singular)
except KeyError:
singular = inflect_engine.singular_noun(field["name"])
if singular is False:
# singular_noun returns False on a noun it can't singularize
singular = field["name"]
return singular
def xproto_singularize_pluralize(field):
try:
# The user has set a plural, as an exception that cannot be handled automatically
plural = field["options"]["plural"]
plural = unquote(plural)
except KeyError:
singular = inflect_engine.singular_noun(field["name"])
if singular is False:
# singular_noun returns False on a noun it can't singularize
singular = field["name"]
plural = inflect_engine.plural_noun(singular)
return plural
def xproto_pluralize(field):
try:
# The user has set a plural, as an exception that cannot be handled automatically
plural = field["options"]["plural"]
plural = unquote(plural)
except KeyError:
plural = inflect_engine.plural_noun(field["name"])
return plural
def xproto_base_def(model_name, base, suffix="", suffix_list=[]):
if model_name == "XOSBase":
return "(models.Model, PlModelMixIn)"
elif not base:
return ""
else:
int_base = [i["name"] + suffix for i in base if i["name"] in suffix_list]
ext_base = [i["name"] for i in base if i["name"] not in suffix_list]
return "(" + ",".join(int_base + ext_base) + ")"
def xproto_first_non_empty(lst):
for l in lst:
if l:
return l
def xproto_api_type(field):
try:
if unquote(field["options"]["content_type"]) == "date":
return "double"
except KeyError:
pass
return field["type"]
def xproto_base_name(n):
# Hack - Refactor NetworkParameter* to make this go away
if n.startswith("NetworkParameter"):
return "_"
expr = r"^[A-Z]+[a-z]*"
try:
match = re.findall(expr, n)[0]
except BaseException:
return "_"
return match
def xproto_base_fields(m, table):
fields = []
for b in m["bases"]:
option1 = b["fqn"]
try:
option2 = m["package"] + "." + b["name"]
except TypeError:
option2 = option1
accessor = None
if option1 in table:
accessor = option1
elif option2 in table:
accessor = option2
if accessor:
base_fields = xproto_base_fields(table[accessor], table)
model_fields = [x.copy() for x in table[accessor]["fields"]]
for field in model_fields:
field["accessor"] = accessor
fields.extend(base_fields)
fields.extend(model_fields)
if "no_sync" in m["options"] and m["options"]["no_sync"]:
fields = [
f
for f in fields
if f["name"] != "backend_status" and f["name"] != "backend_code"
]
if "no_policy" in m["options"] and m["options"]["no_policy"]:
fields = [
f
for f in fields
if f["name"] != "policy_status" and f["name"] != "policy_code"
]
return fields
def xproto_fields(m, table):
""" Generate the full list of models for the xproto message `m` including fields from the classes it inherits.
Inserts the special field "id" at the very beginning.
Each time we descend a new level of inheritance, increment the offset field numbers by 100. The base
class's fields will be numbered from 1-99, the first descendant will be number 100-199, the second
descdendant numbered from 200-299, and so on. This assumes any particular model as at most 100
fields.
"""
model_fields = [x.copy() for x in m["fields"]]
for field in model_fields:
field["accessor"] = m["fqn"]
fields = xproto_base_fields(m, table) + model_fields
# The "id" field is a special field. Every model has one. Put it up front and pretend it's part of the
if not fields:
raise Exception(
"Model %s has no fields. Check for missing base class." % m["name"]
)
id_field = {
"type": "int32",
"name": "id",
"options": {},
"id": "1",
"accessor": fields[0]["accessor"],
}
fields = [id_field] + fields
# Walk through the list of fields. They will be in depth-first search order from the base model forward. Each time
# the model changes, offset the protobuf field numbers by 100.
offset = 0
last_accessor = fields[0]["accessor"]
for field in fields:
if field["accessor"] != last_accessor:
last_accessor = field["accessor"]
offset += 100
field_id = int(field["id"])
if (field_id < 1) or (field_id >= 100):
raise Exception(
"Only field numbers from 1 to 99 are permitted, field %s in model %s"
% (field["name"], field["accessor"])
)
field["id"] = int(field["id"]) + offset
# Check for duplicates
fields_by_number = {}
for field in fields:
id = field["id"]
dup = fields_by_number.get(id)
if dup:
raise Exception(
"Field %s has duplicate number %d with field %s in model %s"
% (field["name"], id, dup["name"], field["accessor"])
)
fields_by_number[id] = field
return fields
def xproto_base_rlinks(m, table):
links = []
for base in m["bases"]:
b = base["name"]
if b in table:
base_rlinks = xproto_base_rlinks(table[b], table)
model_rlinks = [x.copy() for x in table[b]["rlinks"]]
for link in model_rlinks:
link["accessor"] = b
links.extend(base_rlinks)
links.extend(model_rlinks)
return links
def xproto_rlinks(m, table):
""" Return the reverse links for the xproto message `m`.
If the link includes a reverse_id, then it will be used for the protobuf field id. Each level of inheritance
will add an offset of 100 to the supplied reverse_id.
If there is no reverse_id, then one will automatically be allocated started at id 1900. It is encouraged that
all links include reverse_ids, so that field identifiers are deterministic across all protobuf messages.
"""
model_rlinks = [x.copy() for x in m["rlinks"]]
for link in model_rlinks:
link["accessor"] = m["fqn"]
links = xproto_base_rlinks(m, table) + model_rlinks
links = [
x for x in links if ("+" not in x["src_port"]) and ("+" not in x["dst_port"])
]
if links:
last_accessor = links[0]["accessor"]
offset = 0
index = 1900
for link in links:
if link["accessor"] != last_accessor:
last_accessor = link["accessor"]
offset += 100
if link["reverse_id"]:
# Statically numbered reverse links. Use the id that the developer supplied, adding the offset based on
# inheritance depth.
link["id"] = int(link["reverse_id"]) + offset
else:
# Automatically numbered reverse links. These will eventually go away.
link["id"] = index
index += 1
# check for duplicates
links_by_number = {}
for link in links:
id = link["id"]
dup = links_by_number.get(id)
if dup:
raise Exception(
"Field %s has duplicate number %d in model %s with reverse field %s"
% (link["src_port"], id, m["name"], dup["src_port"])
)
links_by_number[id] = link
return links
def xproto_base_links(m, table):
links = []
for base in m["bases"]:
b = base["name"]
if b in table:
base_links = xproto_base_links(table[b], table)
model_links = table[b]["links"]
links.extend(base_links)
links.extend(model_links)
return links
def xproto_string_type(xptags):
try:
max_length = eval(xptags["max_length"])
except BaseException:
max_length = 1024
if "varchar" not in xptags:
return "string"
else:
return "text"
def xproto_tuplify(nested_list_or_set):
if not isinstance(nested_list_or_set, list) and not isinstance(
nested_list_or_set, set
):
return nested_list_or_set
else:
return tuple([xproto_tuplify(i) for i in nested_list_or_set])
def xproto_field_graph_components(fields, model, tag="unique_with"):
def find_components(graph):
pending = set(graph.keys())
components = []
while pending:
front = {pending.pop()}
component = set()
while front:
node = front.pop()
neighbours = graph[node]
neighbours -= component # These we have already visited
front |= neighbours
pending -= neighbours
component |= neighbours
components.append(component)
return components
field_graph = {}
field_names = {f["name"] for f in fields}
for f in fields:
try:
tagged_str = unquote(f["options"][tag])
tagged_fields = tagged_str.split(",")
for uf in tagged_fields:
if uf not in field_names:
raise FieldNotFound(
'Field "%s" not found in model "%s", referenced from field "%s" by option "%s"'
% (uf, model["name"], f["name"], tag)
)
field_graph.setdefault(f["name"], set()).add(uf)
field_graph.setdefault(uf, set()).add(f["name"])
except KeyError:
pass
return find_components(field_graph)
def xproto_api_opts(field):
options = []
if "max_length" in field["options"] and field["type"] == "string":
options.append("(val).maxLength = %s" % field["options"]["max_length"])
try:
if field["options"]["null"] == "False":
options.append("(val).nonNull = true")
except KeyError:
pass
if "link" in field and "model" in field["options"]:
options.append('(foreignKey).modelName = "%s"' % field["options"]["model"])
if ("options" in field) and ("port" in field["options"]):
options.append(
'(foreignKey).reverseFieldName = "%s"' % field["options"]["port"]
)
if options:
options_str = "[" + ", ".join(options) + "]"
else:
options_str = ""
return options_str
def xproto_type_to_swagger_type(f):
try:
content_type = f["options"]["content_type"]
content_type = eval(content_type)
except BaseException:
content_type = None
pass
if "choices" in f["options"]:
return "string"
elif content_type == "date":
return "string"
elif f["type"] == "bool":
return "boolean"
elif f["type"] == "string":
return "string"
elif f["type"] in ["int", "uint32", "int32"] or "link" in f:
return "integer"
elif f["type"] in ["double", "float"]:
return "string"
def xproto_field_to_swagger_enum(f):
if "choices" in f["options"]:
list = []
for c in eval(xproto_unquote(f["options"]["choices"])):
list.append(c[0])
return list
else:
return False
def xproto_is_true(x):
# TODO: Audit xproto and make specification of trueness more uniform
if x is True or (x == "True") or (x == '"True"'):
return True
return False