in aws/lambda/github-webhook-rds-sync/utils.py [0:0]
def generate_orm(name: str, obj: FlatDict, sql_base: Any) -> Any:
"""
Create an instance of a SQLAlchemy ORM class from a dictionary.
"""
columns = {"__tablename__": name, "__table_args__": {"extend_existing": True}}
errors = []
for key, value in obj.items():
col = get_column(key, value, type_name=name)
if col is OBJECT_PLACEHOLDER:
# There is a null object (with a node_id) missing, so create it as
# we would if something was there but leave the value blank
columns[f"{key}_node_id"] = Column(String(100))
elif col is None:
# Unable to find a type for this value. An entry is missing in the
# TYPE_MAP for this name.key pair
errors.append(f"{name} -> {key}: {value}")
else:
# Got a column successfully, so set it on the table
columns[key] = col
if len(errors) > 0:
# Couldn't get a column type for some of the data, so error out
catted_errors = "\n ".join([f"typeerr: {e}" for e in errors])
raise RuntimeError(f"Unknown types:\n{catted_errors}")
# Change data into the right types for storage
obj = transform_data(obj)
# Fill in any inconsistent / missing columns from the GitHub API
# The loop above only looks at the data actually received on the webhook.
# Some things may be missing (inconsistencies in GitHub's API or just
# doesn't exist), so fill in their types here:
for key, column_creator in TYPE_MAP.get(name, {}).items():
value = column_creator()
if value is OBJECT_PLACEHOLDER:
columns[f"{key}_node_id"] = Column(String(50))
if key in obj:
if obj[key] is not None:
raise RuntimeError(f"not doing it {name}.{key}")
else:
del obj[key]
obj[f"{key}_node_id"] = None
else:
columns[key] = value
# Set the primary key (some webhooks don't have a node_id at the top level
# so set up an auto-incrementing int ID for them)
pk_name, pk_type = get_primary_key(name, obj)
columns[pk_name] = pk_type
# Create SQLAlchemy ORM class (which registers it to be created in sql_base)
the_class = type(name, (sql_base,), columns)
return the_class(**obj)