Skip to content
Snippets Groups Projects
Commit cf49668e authored by Florian Spreckelsen's avatar Florian Spreckelsen
Browse files

Merge branch 'f-verbose-option' into 'dev'

ENH: added verbose option to data model

See merge request !28
parents 14a3fbc6 7e6b3903
No related branches found
No related tags found
2 merge requests!39Release 0.4.0,!28ENH: added verbose option to data model
Pipeline #18730 passed
FROM debian:10
FROM debian:11
RUN apt-get update && \
apt-get install \
curl \
......@@ -13,7 +13,7 @@ RUN apt-get update && \
tox \
git \
openjdk-11-jdk-headless \
python-autopep8 \
python3-autopep8 \
python3-pytest \
libxml2 \
-y
......
......@@ -23,6 +23,11 @@
# ** end header
#
from copy import deepcopy
# TODO(fspreck) for backwards compatibility with Python < 3.9 but this is
# actually
# [deprecated](https://docs.python.org/3/library/typing.html#typing.List), so
# remove this, when we drop support for old Python versions.
from typing import List
import caosdb as db
from caosdb.apiutils import compare_entities, describe_diff
......@@ -68,14 +73,14 @@ class DataModel(dict):
else:
super().__init__(args)
def append(self, entity):
def append(self, entity: db.Entity):
self[entity.name] = entity
def extend(self, entities):
def extend(self, entities: List[db.Entity]):
for entity in entities:
self.append(entity)
def sync_data_model(self, noquestion=False):
def sync_data_model(self, noquestion: bool = False, verbose: bool = True):
"""Synchronize this DataModel with a CaosDB instance.
Updates existing entities from the CaosDB instance and inserts
......@@ -100,23 +105,27 @@ class DataModel(dict):
self.sync_ids_by_name(tmp_exist)
if len(non_existing_entities) > 0:
print("New entities:")
if verbose:
print("New entities:")
for ent in non_existing_entities:
print(ent.name)
for ent in non_existing_entities:
print(ent.name)
if noquestion or str(input("Do you really want to insert those "
"entities? [y/N] ")).lower() == "y":
non_existing_entities.insert()
self.sync_ids_by_name(non_existing_entities)
print("Updated entities.")
if verbose:
print("Updated entities.")
else:
return
else:
print("No new entities.")
if verbose:
print("No new entities.")
if len(existing_entities) > 0:
print("Inspecting changes that will be made...")
if verbose:
print("Inspecting changes that will be made...")
any_change = False
for ent in existing_entities:
......@@ -126,18 +135,22 @@ class DataModel(dict):
), name=ent.name))
if diff != "":
print(diff)
if verbose:
print(diff)
any_change = True
if any_change:
if noquestion or input("Do you really want to apply the above "
"changes? [y/N]") == "y":
existing_entities.update()
print("Synchronized existing entities.")
if verbose:
print("Synchronized existing entities.")
else:
print("No differences found. No update")
if verbose:
print("No differences found. No update")
else:
print("No existing entities updated.")
if verbose:
print("No existing entities updated.")
@staticmethod
def get_existing_entities(entities):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment