from collections.abc import AsyncIterator, Iterator import jinja2 import sqlalchemy from anydi import Container, Module, Provider, provider from cachelib import BaseCache as Cache, NullCache, RedisCache, SimpleCache from fansifter_common import context from fansifter_common.adapters.db import jinja2sql_filters from fansifter_common.adapters.graphql_router import GraphqlRouterClient from fansifter_common.adapters.ows_account import OwsAccountClient from fansifter_common.adapters.ows_pdp import OwsPdpClient from fansifter_common.adapters.ows_users import OwsUsersClient from fansifter_common.adapters.sendgrid.client import SendGridClient from fansifter_common.adapters.stripo import StripoClient from fansifter_common.artist.validators import ArtistValidator from fansifter_common.auth.account import Account, AccountAccess from fansifter_common.auth.authorization import ( AccessAuthorizationBackendStub, AuthorizationBackend, PdpAuthorizationBackend, ) from fansifter_common.auth.services import AuthService from fansifter_common.auth.validators import AccountValidator from fansifter_common.constants import QA_ENVIRONMENT from fansifter_common.encrypter import JWTEncrypter from fansifter_common.legal_info.services import LegalInfoService from fansifter_common.translation.services import TranslationService from fansifter_common.utils.functional import lazy_proxy from jinja2sql import Jinja2SQL from jwtauth import JWTAuth from jwtauth.utils import get_default_audience, get_default_issuer, get_default_jwks_url from owsclient import OwsClient from email_campaigns.adapters.db import DB, PgDB from email_campaigns.adapters.features import Features from email_campaigns.adapters.google.client import GoogleFontsClient from email_campaigns.adapters.stripo import ( AutomatedStripoClient, AutomatedStripoClientV2, CampaignStripoClient, CampaignStripoClientV2, ) from email_campaigns.config import Settings, settings class AppModule(Module): @provider(scope="singleton") def jwt_auth(self, settings: Settings) -> JWTAuth: return JWTAuth( jwks_url=get_default_jwks_url(settings.environment), audience=get_default_audience(settings.environment), issuer=get_default_issuer(settings.environment), ) @provider(scope="singleton") def cache(self, settings: Settings) -> Cache: if settings.cache_backend in ["locmem", "memory"]: return SimpleCache(default_timeout=settings.cache_default_timeout) elif settings.cache_backend == "redis": return RedisCache( host=settings.redis_host, port=settings.redis_port, db=settings.cache_redis_db, ssl=settings.redis_ssl, default_timeout=settings.cache_default_timeout, key_prefix=settings.cache_key_prefix, ) return NullCache() @provider(scope="singleton") def ows_client(self, settings: Settings) -> OwsClient: return OwsClient( environment=settings.environment, service_name=settings.service_name, request_context_getter=context.get_request_context, correlation_id_getter=context.get_correlation_id, ) @provider(scope="singleton") def ows_pdp_client(self, ows_client: OwsClient) -> OwsPdpClient: return OwsPdpClient(ows_client=ows_client) @provider(scope="singleton") def ows_account_client(self, ows_client: OwsClient) -> OwsAccountClient: return OwsAccountClient(ows_client=ows_client) @provider(scope="singleton") def ows_users_client(self, ows_client: OwsClient) -> OwsUsersClient: return OwsUsersClient(ows_client=ows_client) @provider(scope="singleton") def graphql_router_client(self, ows_client: OwsClient) -> GraphqlRouterClient: return GraphqlRouterClient(ows_client=ows_client) @provider(scope="singleton") def jinja2sql(self, settings: Settings) -> Jinja2SQL: env = jinja2.Environment( loader=jinja2.FileSystemLoader(settings.jinja2sql_template_searchpath) ) jinja2sql = Jinja2SQL(env) jinja2sql.env.globals.update( # type: ignore { "settings": settings, } ) # Register filters jinja2sql.register_filter( "escape_like", jinja2sql_filters.escape_like, ) jinja2sql.register_filter( "orderby", jinja2sql_filters.orderby_filter, bind=True, ) return jinja2sql @provider(scope="singleton") def db(self, settings: Settings, jinja2sql: Jinja2SQL) -> Iterator[DB]: with DB( url=settings.snowflake_url, engine_args={ "echo": settings.snowflake_echo, "poolclass": sqlalchemy.pool.QueuePool, "pool_size": settings.snowflake_pool_size, "max_overflow": settings.snowflake_pool_max_overflow, "pool_recycle": settings.snowflake_pool_recycle, "pool_pre_ping": settings.snowflake_pool_pre_ping, "pool_reset_on_return": settings.snowflake_pool_reset_on_return, "connect_args": settings.snowflake_connect_args, }, jinja2sql=jinja2sql, ) as db: yield db @provider(scope="request") async def db_session_factory(self, db: DB) -> AsyncIterator[None]: with db.session_factory(): yield @provider(scope="singleton") def pg_db(self, settings: Settings, jinja2sql: Jinja2SQL) -> Iterator[PgDB]: with PgDB( url=settings.postgres_url, engine_args={ "echo": settings.postgres_echo, "poolclass": sqlalchemy.pool.QueuePool, "pool_size": settings.postgres_pool_size, "max_overflow": settings.postgres_pool_max_overflow, "pool_recycle": settings.postgres_pool_recycle, "pool_pre_ping": settings.postgres_pool_pre_ping, "pool_reset_on_return": settings.postgres_pool_reset_on_return, "connect_args": settings.postgres_connect_args, }, jinja2sql=jinja2sql, ) as db: yield db @provider(scope="request") async def pg_db_session_factory(self, db: PgDB) -> AsyncIterator[None]: with db.session_factory(): yield @provider(scope="singleton") def authorization_backend( self, ows_pdp_client: OwsPdpClient, ows_account_client: OwsAccountClient, cache: Cache, settings: Settings, ) -> AuthorizationBackend: # Local/dev environment only if settings.auth_allow_full_access: return AccessAuthorizationBackendStub( AccountAccess( accounts=[ Account(vendor_id=7123, subaccount_id=0), Account(vendor_id=34514, subaccount_id=0), ] ) ) return PdpAuthorizationBackend( ows_pdp_client=ows_pdp_client, ows_account_client=ows_account_client, cache=cache, cache_timeout=settings.account_access_cache_timeout, ) # Services @provider(scope="singleton") def auth_service(self, authorization_backend: AuthorizationBackend) -> AuthService: return AuthService(authorization_backend=authorization_backend) @provider(scope="singleton") def translation_service(self, settings: Settings) -> TranslationService: return TranslationService(translations_path=settings.translations_path) @provider(scope="singleton") def preference_center_encrypter(self, settings: Settings) -> JWTEncrypter: return JWTEncrypter( key=settings.preference_center_secret_key, ) @provider(scope="singleton") def legal_info_service(self) -> LegalInfoService: return LegalInfoService() # Features @provider(scope="singleton") def features(self, settings: Settings) -> Iterator[Features]: features = Features( settings.splitio_api_key, block_until_ready_timeout=settings.splitio_block_until_ready_timeout, config=settings.splitio_config, ) if settings.environment == QA_ENVIRONMENT: features.start() yield features features.close() # API clients @provider(scope="singleton") def google_fonts_client(self, settings: Settings) -> GoogleFontsClient: return GoogleFontsClient(settings=settings) @provider(scope="singleton") def sendgrid_client(self, settings: Settings) -> Iterator[SendGridClient]: with SendGridClient( sme_api_key=settings.sendgrid_sme_api_key, orchard_api_key=settings.sendgrid_orchard_api_key, awal_api_key=settings.sendgrid_awal_api_key, unsubscribe_host=settings.unsubscribe_host, preference_center_url=settings.preference_center_url, reply_to_sme_domain=settings.reply_to_sme_domain, reply_to_orchard_domain=settings.reply_to_orchard_domain, reply_to_awal_domain=settings.reply_to_awal_domain, reply_to_header_enabled=settings.reply_to_header_enabled, reply_to_secret_key=settings.reply_to_secret_key.get_secret_value(), ) as sendgrid_client: yield sendgrid_client @provider(scope="singleton") def stripo_client(self, settings: Settings) -> Iterator[StripoClient]: with StripoClient( base_url=settings.stripo_base_url, client_id=settings.stripo_client_id, client_secret=settings.stripo_client_secret, ) as stripo_client: yield stripo_client @provider(scope="singleton") def campaign_stripo_client( self, settings: Settings ) -> Iterator[CampaignStripoClient]: with CampaignStripoClient( base_url=settings.stripo_base_url, client_id=settings.stripo_client_id, client_secret=settings.stripo_client_secret, ) as client: yield client @provider(scope="singleton") def automated_stripo_client( self, settings: Settings ) -> Iterator[AutomatedStripoClient]: with AutomatedStripoClient( base_url=settings.stripo_base_url, client_id=settings.stripo_automated_client_id, client_secret=settings.stripo_automated_client_secret, ) as client: yield client @provider(scope="singleton") def campaign_stripo_client_v2( self, settings: Settings ) -> Iterator[CampaignStripoClientV2]: with CampaignStripoClientV2( base_url=settings.stripo_v2_base_url, coediting_url=settings.stripo_v2_coediting_url, client_id=settings.stripo_v2_client_id, client_secret=settings.stripo_v2_client_secret, ) as client: yield client @provider(scope="singleton") def automated_stripo_client_v2( self, settings: Settings ) -> Iterator[AutomatedStripoClientV2]: with AutomatedStripoClientV2( base_url=settings.stripo_v2_base_url, coediting_url=settings.stripo_v2_coediting_url, client_id=settings.stripo_v2_automated_client_id, client_secret=settings.stripo_v2_automated_client_secret, ) as client: yield client # Validators @provider(scope="singleton") def artist_validator( self, graphql_router_client: GraphqlRouterClient ) -> ArtistValidator: return ArtistValidator(graphql_router_client=graphql_router_client) @provider(scope="singleton") def account_validator( self, ows_account_client: OwsAccountClient ) -> AccountValidator: return AccountValidator(ows_account_client=ows_account_client) def make_container() -> Container: """Configure the application.""" container = Container( providers=[ Provider(Settings, factory=lambda: settings, scope="singleton"), ], modules=[AppModule], ) container.scan( "email_campaigns", ignore=[".container", ".adapters", ".api", ".cli", ".utils"], ) container.build() return container container = lazy_proxy(make_container)