"""Configuration handler for the API. The philosophies for howRootski handles configuration come from the "12-Factor App".All configuration values can come from the following sources, and willbe prioritized in the following order:1. CLI arguments (unused in this app)2. Environment variables3. Config file4. Default values.. note:: If the name of a config value in the YAML file is "name", then the environment variable equivalent will be ROOTSKI__NAME."""importjsonimportosfromenumimportEnumfrompathlibimportPathfromtypingimportAny,Dict,List,Tuple,UnionimportyamlfrompydanticimportAnyHttpUrl,BaseSettings,validatorfrompydantic.dataclassesimportdataclassfrompydantic.env_settingsimportSettingsSourceCallablefromrootski.config.ssmimportget_ssm_parameters_by_prefixANON_USER="anon@rootski.io"ENVIRON_PREFIX:str="ROOTSKI__"# default config values#: default port for the FastAPI server to listen onDEFAULT_PORT:int=3333#: default host for the FastAPI server to listen onDEFAULT_HOST:str="0.0.0.0"#: default domain for the API (used for CORS and may be for other things)DEFAULT_DOMAIN:str="www.rootski.io"#: default http:// URL of the S3 bucket containing the static frontend filesDEFAULT_S3_STATIC_SITE_DOMAIN="http://io.rootski.www.s3-website-us-west-2.amazonaws.com"#: environment variable where the rootski API config file should be foundYAML_CONFIG_PATH_ENV_VAR:str=f"{ENVIRON_PREFIX}CONFIG_FILE_PATH"# we expect this to be one of "dev" or "prod"DEPLOYMENT_ENVIRONMENT_ENV_VAR=f"{ENVIRON_PREFIX}ENVIRONMENT"DEFAULT_DEPLOYMENT_ENVIRONMENT="dev"DEFAULT_DYNAMO_TABLE_NAME="rootski-table"# maps to a string booleanFETCH_VALUES_FROM_SSM_ENV_VAR=f"{ENVIRON_PREFIX}FETCH_VALUES_FROM_AWS_SSM"########################################################## --- Helper functions to set Rootski config values --- ##########################################################
[docs]defget_environ_name(name:str)->str:"""Get the environment variable name for a given config value."""returnf"{ENVIRON_PREFIX}{name.upper()}"
[docs]defload_config_from_yaml(config_fpath:str)->Dict[str,str]:"""Read config from a YAML file."""withPath(config_fpath).open("r")asf:config=yaml.safe_load(f)returnconfig
[docs]defyaml_config_settings_source(settings:"Config")->Dict[str,Any]:"""App settings from the yaml config file."""yaml_config_path=os.environ.get(YAML_CONFIG_PATH_ENV_VAR)ifnotyaml_config_path:return{}returnload_config_from_yaml(config_fpath=yaml_config_path)ifyaml_config_pathelse{}
[docs]defaws_parameter_store_settings_source(settings:"Config")->Dict[str,Any]:""" Fetch app settings from AWS SSM Parameter Store. If a previous settings provider has set the ``fetch_values_from_ssm`` value to ``False``, this function will not attempt to fetch values from SSM. """fetch_values_from_aws_ssm:bool=os.environ.get(FETCH_VALUES_FROM_SSM_ENV_VAR,"false").lower()=="true"ifnotfetch_values_from_aws_ssm:return{}deployment_environment=os.environ.get(DEPLOYMENT_ENVIRONMENT_ENV_VAR,DEFAULT_DEPLOYMENT_ENVIRONMENT)rootski_params:Dict[str,str]=get_ssm_parameters_by_prefix(prefix=f"/rootski/{deployment_environment}/")to_return:Dict[str,str]={}if"database_config"inrootski_params.keys():database_config=json.loads(rootski_params["database_config"])to_return.update({"postgres_user":database_config["postgres_user"],"postgres_password":database_config["postgres_password"],"postgres_host":database_config["postgres_host"],"postgres_port":database_config["postgres_port"],"postgres_db":database_config["postgres_db"],})defupdate__to_return__if_key_present_in__rootski_params(key:str):ifkeyinrootski_params.keys():to_return[key]=rootski_params[key]forkeyin["cognito_aws_region","cognito_user_pool_id","cognito_web_client_id"]:update__to_return__if_key_present_in__rootski_params(key)returnto_return
[docs]defdefault_settings(settings:"Config")->Dict[str,Any]:"""Default values for certain app settings."""# these keys must be named the same as the attributes in the# Config class belowreturn{# "s3_static_site_origin": DEFAULT_S3_STATIC_SITE_DOMAIN,"host":DEFAULT_HOST,"port":DEFAULT_PORT,"domain":DEFAULT_DOMAIN,"extra_allowed_cors_origins":[],}
[docs]@dataclass(frozen=True)classConfig(BaseSettings):"""A configuration manager for the app. Configuration values are discovered and kept in the following order of priority: 1. Init arguments (could be used to give CLI arguments highest priority) 2. Environment variables of the form ``<ENVIRON_PREFIX>UPPER_CASE_ATTRIBUTE`` 3. Config yaml file, whose location is at ``<ENVIRON_PREFIX>CONFIG_FILE_PATH`` 4. Default values set in this ``dataclass`` .. note:: When using environment variables, values should be JSON formatted strings. For example, ``"link1,link2,link3"`` would fail for ``extra_allowed_cors_origins``, but ``'["link1", "link2", "link3"]'`` would work. .. note:: This ``__init__()`` method only exists so that there is a class-level docstring in the sphinx documentation. """log_level:LogLevel=LogLevel.INFO.valuehost:str=DEFAULT_HOSTport:int=DEFAULT_PORTdomain:str=DEFAULT_DOMAINs3_static_site_origin:str=DEFAULT_S3_STATIC_SITE_DOMAINcognito_aws_region:strcognito_user_pool_id:strcognito_web_client_id:strstatic_assets_dir:str=str((Path(__file__).parent/"../../../static").resolve())extra_allowed_cors_origins:List[AnyHttpUrl]=[]dynamo_table_name:str=DEFAULT_DYNAMO_TABLE_NAME@propertydefstatic_morphemes_json_fpath(self)->Path:returnPath(self.static_assets_dir)/"morphemes.json"# TODO - uncomment these; unfortunately, as of Nov 1, 2021, pydantic does not support# "postgressql+psycopg2" or "postgresql+asyncpg" as schemas for the PostgresDsn. This# is coming in the next release, but when I installed the latest release from GitHub# many other portions of the app broke, so for now, these will be constructed using# properties# @validator("sync_sqlalchemy_database_uri", pre=True)# def assemble_sync_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> str:# """Simultaneously validate db connection params and build the connection string."""# if isinstance(v, str):# return v# return PostgresDsn.build(# scheme="postgresql+psycopg2",# user=values.get("postgres_user"),# password=values.get("postgres_password"),# host=values.get("postgres_host"),# port=values.get("postgres_port"),# path=f"/{values.get('postgres_db') or ''}",# )# @validator("async_sqlalchemy_database_uri", pre=True)# def assemble_async_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> str:# """Simultaneously validate db connection params and build the connection string."""# if isinstance(v, str):# return v# return PostgresDsn.build(# scheme="postgresql+asyncpg",# user=values.get("postgres_user"),# password=values.get("postgres_password"),# host=values.get("postgres_host"),# port=values.get("postgres_port"),# path=f"/{values.get('postgres_db') or ''}",# )@propertydefallowed_cors_origins(self)->List[AnyHttpUrl]:returnself.extra_allowed_cors_origins+[self.s3_static_site_origin,f"http://localhost:{self.port}",f"https://localhost:{self.port}",f"https://{self.domain}",]@propertydefcognito_public_keys_url(self)->str:"""URL of Cognito public keys used to validate cognito issued JWT tokens for the Rootski user pool."""returnf"https://cognito-idp.{self.cognito_aws_region}.amazonaws.com/{self.cognito_user_pool_id}/.well-known/jwks.json"
[docs]@validator("extra_allowed_cors_origins",pre=True)defassemble_cors_origins(cls,v:Union[str,List[str]])->Union[List[str],str]:"""Ensure ``extra_allowed_cors_origins`` is a valid list of strings. Parse it from a str to a List[str] if needed."""# parse comma separated strings to List[str]ifisinstance(v,str)andnotv.startswith("["):return[i.strip()foriinv.split(",")]elifisinstance(v,(list,str)):returnvraiseValueError(v)
classConfig:env_prefix=ENVIRON_PREFIXcase_sensitive=Falseuse_enum_values=True@classmethoddefcustomise_sources(cls,init_settings:SettingsSourceCallable,env_settings:SettingsSourceCallable,file_secret_settings:SettingsSourceCallable,)->Tuple[SettingsSourceCallable,...]:"""Register settings providers. Config providers will be prioritized in the order they are returned here."""return(init_settings,# kwargs to Config() constructorenv_settings,# environment variable versions of config valuesyaml_config_settings_source,# values from yaml fileaws_parameter_store_settings_source,# values from ssm parameter storefile_secret_settings,# ???)def__init__(self,**kwargs):# The tests seem to fail without this empty __init__# pylint: disable=useless-super-delegationsuper().__init__(**kwargs)