Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
)
# creating the users takes half of the time
progress = (100 / self.number_lines) * 0.5
for user in users:
self.progress_update(progress)
user_row = users[user]
values = self.get_field_values(user_row)
if values["uuid"] in existing_users:
user_obj = FacilityUser.objects.get(
id=values["uuid"], facility=self.default_facility
)
keeping_users.append(user_obj)
if user_obj.username != user:
# check for duplicated username in the facility
existing_user = FacilityUser.objects.get(
username=user, facility=self.default_facility
)
if existing_user:
error = {
"row": users[user]["position"],
"username": user,
"message": MESSAGES[DUPLICATED_USERNAME],
"field": "USERNAME",
"value": user,
}
per_line_errors.append(error)
continue
if self.compare_fields(user_obj, values):
update_users.append(user_obj)
else:
if values["uuid"] != "":
def has_morango_certificate_scope_permission(
self, scope_definition_id, scope_params
):
if self.is_superuser:
# superusers of a device always have permission to sync
return True
if scope_params.get("dataset_id") != self.dataset_id:
# if the request isn't for the same facility as this user, abort
return False
if scope_definition_id == ScopeDefinitions.FULL_FACILITY:
# if request is for full-facility syncing, return True only if user is a Facility Admin
return self.has_role_for_collection(role_kinds.ADMIN, self.facility)
elif scope_definition_id == ScopeDefinitions.SINGLE_USER:
# for single-user syncing, return True if this user *is* target user, or is admin for target user
target_user = FacilityUser.objects.get(id=scope_params.get("user_id"))
if self == target_user:
return True
if self.has_role_for_user(target_user, role_kinds.ADMIN):
return True
return False
return False
if u[self.header_translation["UUID"]] != ""
]
existing_users = (
FacilityUser.objects.filter(facility=self.default_facility)
.filter(id__in=users_uuid)
.values_list("id", flat=True)
)
# creating the users takes half of the time
progress = (100 / self.number_lines) * 0.5
for user in users:
self.progress_update(progress)
user_row = users[user]
values = self.get_field_values(user_row)
if values["uuid"] in existing_users:
user_obj = FacilityUser.objects.get(
id=values["uuid"], facility=self.default_facility
)
keeping_users.append(user_obj)
if user_obj.username != user:
# check for duplicated username in the facility
existing_user = FacilityUser.objects.get(
username=user, facility=self.default_facility
)
if existing_user:
error = {
"row": users[user]["position"],
"username": user,
"message": MESSAGES[DUPLICATED_USERNAME],
"field": "USERNAME",
"value": user,
}
def get_user(self, username, users):
user = users.get(username, None)
if not user: # the user has not been created nor updated:
user = FacilityUser.objects.get(
username=username, facility=self.default_facility
)
return user
def get_user_name(user_id):
try:
user = FacilityUser.objects.get(pk=user_id)
return user.full_name
except FacilityUser.DoesNotExist:
return ''
def validate(self, attrs):
username = attrs.get("username")
# first condition is for creating object, second is for updating
facility = attrs.get("facility") or getattr(self.instance, "facility")
# if obj doesn't exist, return data
try:
obj = FacilityUser.objects.get(username__iexact=username, facility=facility)
except FacilityUser.DoesNotExist:
return attrs
# if we are updating object, and this `instance` is the same object, return data
if self.instance and obj.id == self.instance.id:
return attrs
else:
raise serializers.ValidationError(
"An account with that username already exists.",
code=error_constants.USERNAME_ALREADY_EXISTS,
)
def create_user(i, user, default_facility=None):
validate_username(user)
if i == 0 and all(key == val or val is None for key, val in user.items()):
# Check whether the first row is a header row or not
# Either each key will be equal to the value
# Or the header is not included in the CSV, so it is None
return False
facility = infer_facility(user, default_facility)
classroom = infer_and_create_class(user, facility)
username = user["username"]
try:
user_obj = FacilityUser.objects.get(username=username, facility=facility)
logger.warn(
"Tried to create a user with the username {username} in facility {facility}, but one already exists".format(
username=username, facility=facility
)
)
if classroom:
classroom.add_member(user_obj)
return False
except FacilityUser.DoesNotExist:
password = user.get("password", DEFAULT_PASSWORD) or DEFAULT_PASSWORD
try:
new_user = FacilityUser.objects.create_user(
full_name=user.get("full_name", ""),
username=username,
facility=facility,
password=password,