pyceo/ceo/cli/database.py

71 lines
2.5 KiB
Python

import os
from typing import Dict
import click
from zope import component
from ..utils import http_post, http_get, http_delete, write_db_creds
from .utils import handle_sync_response, 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')
if db_type == 'mysql':
db_type_name = 'MySQL'
else:
db_type_name = 'PostgreSQL'
wrote_to_file = write_db_creds(filename, user_dict, password, db_type, db_host)
if op == 'create':
click.echo(f'{db_type_name} database created.')
username = user_dict['uid']
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")
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")
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)