|
import argparse |
|
import logging |
|
import os |
|
|
|
import faculty |
|
|
|
# Roles to give to users as they are added |
|
ROLES = ["project-data-scientist", "project-observer", "project-data-provider"] |
|
|
|
|
|
DESCRIPTION = """\ |
|
Add users to a Faculty project in bulk. |
|
|
|
This script reads a file containing emails of users, with one email per line. |
|
It then adds these users as data scientists in the current project, provided |
|
the users exist in the platform. |
|
""" |
|
|
|
|
|
def parse_arguments(): |
|
parser = argparse.ArgumentParser(description=DESCRIPTION) |
|
parser.add_argument( |
|
"email_file", |
|
type=argparse.FileType(mode="r"), |
|
help="file containing emails to add, with one email per file." |
|
) |
|
|
|
parser.add_argument("-v", "--verbose", action="store_true", help="increase verbosity") |
|
args = parser.parse_args() |
|
return args |
|
|
|
|
|
def get_users_for_emails(user_emails): |
|
user_client = faculty.client("user") |
|
all_users = user_client.get_all_users(is_system=False, enabled=True) |
|
email_to_user = {user.email: user for user in all_users} |
|
found_users = [email_to_user[email] for email in user_emails if email in email_to_user] |
|
missing_emails = [email for email in user_emails if email not in email_to_user] |
|
return found_users, missing_emails |
|
|
|
|
|
def add_user_to_project(user_id, project_id): |
|
project_client = faculty.client("project") |
|
response = project_client._post_raw(f"/project/{project_id}/member/{user_id}", json={"roles": ROLES}) |
|
response.raise_for_status() |
|
|
|
|
|
def main(email_file, project_id): |
|
emails = [line.strip() for line in email_file] |
|
logging.info(f"Found {len(emails)} emails in input file.") |
|
found_users, missing_emails = get_users_for_emails(emails) |
|
logging.info(f"Out of the emails in the input, {len(found_users)} were in Faculty database.") |
|
for email in missing_emails: |
|
logging.warn(f"No user with email {email} found in Faculty database.") |
|
for user in found_users: |
|
add_user_to_project(user.id, project_id) |
|
logging.info(f"Added user {user.email} to project {project_id}.") |
|
|
|
|
|
if __name__ == "__main__": |
|
args = parse_arguments() |
|
if args.verbose: |
|
logging.basicConfig(level=logging.INFO) |
|
project_id = os.getenv("FACULTY_PROJECT_ID") |
|
if project_id is None: |
|
logging.error( |
|
"Missing environment variable FACULTY_PROJECT_ID. " |
|
"This script should be run in the project you want to add users to." |
|
) |
|
else: |
|
logging.info(f"Adding users to project {project_id}.") |
|
logging.info(f"Using {args.email_file.name} as input.") |
|
main(args.email_file, project_id) |