Added: [untested] read/write email config api routes (#284)

Author: signebedi

libreforms_fastapi/app/__init__.py

@@ -119,6 +119,7 @@
     RelationshipTypeModel,
     UserRelationshipModel,
     FormConfigUpdateRequest,
+    EmailConfigUpdateRequest,
     SiteConfig,
     get_user_model,
     get_form_model,
@@ -139,7 +140,12 @@
 from libreforms_fastapi.utils.custom_tinydb import CustomEncoder
 
 # Import to render more powerful email content
-from libreforms_fastapi.utils.jinja_emails import render_email_message_from_jinja
+from libreforms_fastapi.utils.jinja_emails import ( 
+    render_email_message_from_jinja,
+    write_email_config_yaml,
+    test_email_config,
+    get_email_yaml,
+)
 
 # Here we set the application config using the get_config
 # factory pattern defined in libreforms_fastapi.utis.config.
@@ -4552,7 +4558,7 @@ async def api_admin_get_form_config(
         raise HTTPException(status_code=404)
 
 
-    _form_config = _get_form_config_yaml(config_path=config.FORM_CONFIG_PATH)
+    _form_config = get_form_config_yaml(config_path=config.FORM_CONFIG_PATH)
 
     # Write this query to the TransactionLog
     if config.COLLECT_USAGE_STATISTICS:
@@ -4600,7 +4606,7 @@ async def api_admin_write_form_config(
         raise HTTPException(status_code=404)
 
 
-    old_form_config_str = get_form_config_yaml(config_path=config.FORM_CONFIG_PATH).strip()
+    # old_form_config_str = get_form_config_yaml(config_path=config.FORM_CONFIG_PATH).strip()
 
     try:
         write_form_config_yaml(
@@ -4634,6 +4640,115 @@ async def api_admin_write_form_config(
 
 
 
+# Get email yaml
+@app.get(
+    "/api/admin/get_email_config", 
+    dependencies=[Depends(api_key_auth)], 
+    response_class=JSONResponse, 
+)
+async def api_admin_get_email_config(
+    request: Request, 
+    background_tasks: BackgroundTasks,
+    config = Depends(get_config_depends),
+    mailer = Depends(get_mailer), 
+    session: SessionLocal = Depends(get_db), 
+    key: str = Depends(X_API_KEY),
+    return_as_yaml_str: bool = False,
+):
+    """
+    Allows site administrators to view the site email config as yaml. This operation is logged for audit purposes. Set
+    `return_as_yaml_str` to True to receive back a string of the yaml config file.
+    """
+
+    # Get the requesting user details
+    user = session.query(User).filter_by(api_key=key).first()
+
+    if not user or not user.site_admin:
+        raise HTTPException(status_code=404)
+
+    email_config = get_email_yaml(config_path=config.EMAIL_CONFIG_PATH, return_as_yaml_str=return_as_yaml_str)
+
+    # Write this query to the TransactionLog
+    if config.COLLECT_USAGE_STATISTICS:
+
+        endpoint = request.url.path
+        remote_addr = request.client.host
+
+        background_tasks.add_task(
+            write_api_call_to_transaction_log, 
+            api_key=key, 
+            endpoint=endpoint, 
+            remote_addr=remote_addr, 
+            query_params={},
+        )
+
+    return JSONResponse(
+        status_code=200,
+        content={"status": "success", "content": email_config},
+    )
+
+
+# Update email config string
+@app.post(
+    "/api/admin/write_email_config", 
+    dependencies=[Depends(api_key_auth)], 
+    response_class=JSONResponse, 
+)
+async def api_admin_write_email_config(
+    request: Request, 
+    _email_config: EmailConfigUpdateRequest,
+    background_tasks: BackgroundTasks,
+    config = Depends(get_config_depends),
+    mailer = Depends(get_mailer), 
+    session: SessionLocal = Depends(get_db), 
+    key: str = Depends(X_API_KEY)
+):
+    """
+    Allows site administrators to update the site email config as yaml. This operation is logged for audit purposes.
+    """
+
+    # Get the requesting user details
+    user = session.query(User).filter_by(api_key=key).first()
+
+    if not user or not user.site_admin:
+        raise HTTPException(status_code=404)
+
+    # old_config_str = get_email_yaml(config_path=config.EMAIL_CONFIG_PATH, return_as_yaml_str=return_as_yaml_str).strip()
+
+
+    try:
+        write_email_config_yaml(
+            config_path=config.EMAIL_CONFIG_PATH, 
+            email_config_str=_email_config.content, 
+            env=config.ENVIRONMENT,
+            timezone=config.TIMEZONE,
+            config=config, user=user, # Add'l kwargs to validate with better data
+        )
+    except Exception as e:
+        raise HTTPException(status_code=422, detail=f"{e}")
+
+    # Write this query to the TransactionLog
+    if config.COLLECT_USAGE_STATISTICS:
+
+        endpoint = request.url.path
+        remote_addr = request.client.host
+
+        background_tasks.add_task(
+            write_api_call_to_transaction_log, 
+            api_key=key, 
+            endpoint=endpoint, 
+            remote_addr=remote_addr, 
+            query_params={},
+        )
+
+    return JSONResponse(
+        status_code=200,
+        content={"status": "success"},
+    )
+
+
+
+
 # Update form config string
 @app.post(
     "/api/admin/update_site_config", 

libreforms_fastapi/utils/jinja_emails.py

@@ -1,4 +1,5 @@
 import yaml
+from zoneinfo import ZoneInfo
 from jinja2 import Template, Undefined, Environment, make_logging_undefined, select_autoescape
 
 
@@ -117,8 +118,7 @@ def __str__(self):
 
 
 
-
-def get_message_jinja(message_type, config_path):
+def get_email_yaml(config_path, return_as_yaml_str=False):
 
     # We want to add logic to read and overwrite 
     # these template key-value pairs using yaml.
@@ -127,18 +127,35 @@ def get_message_jinja(message_type, config_path):
     try:
         assert(os.path.exists(config_path)) # If it doesn't exist, let's skip this rigamarole
         with open(config_path, 'r') as file:
-            loaded_config = yaml.safe_load(file.read())
+
+            yaml_str = file.read()
+
+        if return_as_yaml_str: 
+            return yaml_str
+
+        loaded_config = yaml.safe_load(yaml_str)
 
     except Exception as e:
+
+        if return_as_yaml_str: 
+            return EXAMPLE_EMAIL_CONFIG_YAML
+
         loaded_config = {}
 
     # Overwrite default config with values from the default path
     for key, value in loaded_config.items():
         default_config[key] = value
 
+    return default_config
+
+
+def get_message_jinja(message_type, config_path):
+
+    config = get_email_yaml(config_path)
+    
     # Retrieve the unrendered Jinja templates from the dictionary
-    subj_template_str = default_config[message_type]['subj']
-    cont_template_str = default_config[message_type]['cont']
+    subj_template_str = config[message_type]['subj']
+    cont_template_str = config[message_type]['cont']
 
     return subj_template_str, cont_template_str
 
@@ -157,6 +174,51 @@ def render_email_message_from_jinja(message_type, config_path, **kwargs):
 
     return rendered_subj, rendered_cont
 
+def write_email_config_yaml(
+    config_path:str,
+    email_config_str:str,
+    env:str,
+    timezone=ZoneInfo("America/New_York"),
+    test_on_write:bool=True,
+    **kwargs,
+):
+    """
+    Here we write the string representation of the yaml email config.
+    """
+
+    if test_on_write:
+        # Attempt to load the YAML string to check its validity
+        _ = test_email_config(email_config_str, **kwargs)
+
+    # Ensure the base directory exists
+    basedir = os.path.dirname(config_path)
+    if not os.path.exists(basedir):
+        os.makedirs(basedir)
+
+    # Create a backup of the current config
+    config_backup_directory = Path(os.getcwd()) / 'instance' / f'{env}_email_config_backups'
+    config_backup_directory.mkdir(parents=True, exist_ok=True)
+
+    datetime_format = datetime.now(timezone).strftime("%Y%m%d%H%M%S")
+    config_file_name = Path(config_path).name
+    backup_file_name = f"{config_file_name}.{datetime_format}"
+    backup_file_path = config_backup_directory / backup_file_name
+
+    # Copy the existing config file to the backup location
+    if os.path.exists(config_path):
+        shutil.copy(config_path, backup_file_path)
+
+    # Write the cleaned YAML string to the config file
+    try:
+        with open(config_path, 'w') as file:
+            # file.write(cleaned_form_config_str)
+            file.write(email_config_str)
+    except Exception as e:
+        raise Exception(f"Failed to write the email config to {config_path}: {e}")
+
+    return True
+
+
 
 def test_email_config(email_config_yaml, **kwargs):
     # Load the YAML configuration into a Python dictionary

libreforms_fastapi/utils/pydantic_models.py

@@ -25,6 +25,12 @@
     get_custom_loader,
 )
 
+
+# from libreforms_fastapi.utils.jinja_emails import (
+#     test_email_config,
+# )
+
+
 class ImproperUsernameFormat(Exception):
     """Raised when the username does not meet the regular expression defined in the app config"""
     pass
@@ -982,6 +988,29 @@ def validate_yaml(cls, v):
             raise ValueError(f"An error occurred while parsing YAML: {e}")
 
 
+class EmailConfigUpdateRequest(BaseModel):
+    """Another quick model for managing admin update email config API calls"""
+    content: str
+
+    @validator('content')
+    def validate_yaml(cls, v):
+
+        try:
+            # Remove leading and trailing double and single quotes
+            v = v.strip('"\'')
+
+            # _ = test_email_config(v, **kwargs)
+
+            data = yaml.safe_load(v)
+
+            return v
+        except yaml.YAMLError as e:
+            raise ValueError(f"The content is not valid YAML: {e}")
+        except Exception as e:
+            raise ValueError(f"An error occurred while parsing YAML: {e}")
+
+
+
 class GroupModel(BaseModel):
     """This model will be used for validating change to Groups through the admin API"""
     # id: int = Field(None)