105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
import os
|
|
from typing import Dict
|
|
|
|
import click
|
|
from zope import component
|
|
|
|
from ..utils import http_post, http_get, http_delete
|
|
from .utils import handle_sync_response, check_file_path, check_if_in_development
|
|
from ceo_common.interfaces import IConfig
|
|
|
|
|
|
def db_cli_response(filename: str, user_dict: Dict, password: str, db_type: str, op: str):
|
|
cfg_srv = component.getUtility(IConfig)
|
|
db_host = cfg_srv.get(f'{db_type}_host')
|
|
username = user_dict['uid']
|
|
if db_type == 'mysql':
|
|
db_type_name = 'MySQL'
|
|
db_cli_local_cmd = f'mysql {username}'
|
|
db_cli_cmd = f'mysql {username} -h {db_host} -u {username} -p'
|
|
else:
|
|
db_type_name = 'PostgreSQL'
|
|
db_cli_local_cmd = f'psql {username}'
|
|
db_cli_cmd = f'psql -d {username} -h {db_host} -U {username} -W'
|
|
username = user_dict['uid']
|
|
info = f"""{db_type_name} Database Information for {username}
|
|
|
|
Your new {db_type_name} database was created. To connect, use the following options:
|
|
|
|
Database: {username}
|
|
Username: {username}
|
|
Password: {password}
|
|
Host: {db_host}
|
|
|
|
On {db_host} to connect using the {db_type_name} command-line client use
|
|
|
|
{db_cli_local_cmd}
|
|
|
|
From other CSC machines you can connect using
|
|
|
|
{db_cli_cmd}
|
|
"""
|
|
wrote_to_file = False
|
|
try:
|
|
# TODO: use phosphoric-acid to write to file (phosphoric-acid makes
|
|
# internal API call to caffeine)
|
|
with click.open_file(filename, "w") as f:
|
|
f.write(info)
|
|
os.chown(filename, user_dict['uid_number'], user_dict['gid_number'])
|
|
os.chmod(filename, 0o640)
|
|
wrote_to_file = True
|
|
except PermissionError:
|
|
pass
|
|
if op == 'create':
|
|
click.echo(f'{db_type_name} database created.')
|
|
click.echo(f'''Connection Information:
|
|
|
|
Database: {username}
|
|
Username: {username}
|
|
Password: {password}
|
|
Host: {db_host}''')
|
|
if wrote_to_file:
|
|
click.echo(f"\nThese settings have been written to {filename}.")
|
|
else:
|
|
click.echo(f"\nWe were unable to write these settings to {filename}.")
|
|
|
|
|
|
def create(username: str, db_type: str):
|
|
db_type_name = 'MySQL' if db_type == 'mysql' else 'PostgreSQL'
|
|
resp = http_get(f'/api/members/{username}')
|
|
user_dict = handle_sync_response(resp)
|
|
click.confirm(f'Are you sure you want to create a {db_type_name} database for {username}?', abort=True)
|
|
|
|
info_file_path = os.path.join(user_dict['home_directory'], f"ceo-{db_type}-info")
|
|
check_file_path(info_file_path)
|
|
|
|
resp = http_post(f'/api/db/{db_type}/{username}')
|
|
result = handle_sync_response(resp)
|
|
password = result['password']
|
|
|
|
db_cli_response(info_file_path, user_dict, password, db_type, 'create')
|
|
|
|
|
|
def pwreset(username: str, db_type: str):
|
|
db_type_name = 'MySQL' if db_type == 'mysql' else 'PostgreSQL'
|
|
resp = http_get(f'/api/members/{username}')
|
|
user_dict = handle_sync_response(resp)
|
|
click.confirm(f'Are you sure you want reset the {db_type_name} password for {username}?', abort=True)
|
|
|
|
info_file_path = os.path.join(user_dict['home_directory'], f"ceo-{db_type}-info")
|
|
check_file_path(info_file_path)
|
|
|
|
resp = http_post(f'/api/db/{db_type}/{username}/pwreset')
|
|
result = handle_sync_response(resp)
|
|
password = result['password']
|
|
|
|
db_cli_response(info_file_path, user_dict, password, db_type, 'pwreset')
|
|
|
|
|
|
def delete(username: str, db_type: str):
|
|
check_if_in_development()
|
|
db_type_name = 'MySQL' if db_type == 'mysql' else 'PostgreSQL'
|
|
click.confirm(f"Are you sure you want to delete the {db_type_name} database for {username}?", abort=True)
|
|
resp = http_delete(f'/api/db/{db_type}/{username}')
|
|
handle_sync_response(resp)
|