app.py

import os
from datetime import timedelta, date, datetime

import uvicorn
from apscheduler.triggers.interval import IntervalTrigger
from environs import Env
from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware as WSGIMiddlewareFastAPI
from flask import Flask
from flask.json.provider import DefaultJSONProvider
from flask_cors import CORS
from flask_restx import Api
from starlette.applications import Starlette
from starlette.middleware.cors import CORSMiddleware

from config import config, oauth, limiter, jwt
from middleware.access_log_filter import (
    SuppressTestAccessLogMiddleware,
    install_test_request_log_filter,
)
from db.helpers_.psycopg import initialize_psycopg_connection
from endpoints.instantiations.admin_.routes import namespace_admin
from endpoints.instantiations.agencies_.routes import namespace_agencies
from endpoints.instantiations.auth_.callback import namespace_callback
from endpoints.instantiations.auth_.login import namespace_login
from endpoints.instantiations.auth_.refresh_session import namespace_refresh_session
from endpoints.instantiations.auth_.request_reset_password import (
    namespace_request_reset_password,
)
from endpoints.instantiations.auth_.resend_validation_email.route import (
    namespace_resend_validation_email,
)
from endpoints.instantiations.auth_.reset_password import namespace_reset_password
from endpoints.instantiations.auth_.reset_token_validation import (
    namespace_reset_token_validation,
)
from endpoints.instantiations.auth_.routes import namespace_auth
from endpoints.instantiations.auth_.signup.route import namespace_signup
from endpoints.instantiations.auth_.validate_email.route import namespace_validate_email
from endpoints.instantiations.check_.route import namespace_url_checker
from endpoints.instantiations.contact_.route import namespace_contact
from endpoints.instantiations.data_requests_.data_requests import (
    namespace_data_requests,
)
from endpoints.instantiations.data_sources_.routes import namespace_data_source
from endpoints.instantiations.dev_.route import (
    namespace_create_test_user,
)
from endpoints.instantiations.github_.route import namespace_github
from endpoints.instantiations.locations_.locations import namespace_locations
from endpoints.instantiations.map.routes import namespace_map
from endpoints.instantiations.match_.route import namespace_match
from endpoints.instantiations.metadata_.route import namespace_metadata
from endpoints.instantiations.metrics_.metrics import namespace_metrics
from endpoints.instantiations.notifications_.route import namespace_notifications
from endpoints.instantiations.oauth_.link_to_github import namespace_link_to_github
from endpoints.instantiations.oauth_.login_with_github import (
    namespace_login_with_github,
)
from endpoints.instantiations.oauth_.oauth import namespace_oauth
from endpoints.instantiations.search.routes import namespace_search
from endpoints.instantiations.source_collector.routes import (
    namespace_source_collector,
)
from endpoints.instantiations.typeahead_.routes import (
    namespace_typeahead_suggestions,
)
from endpoints.instantiations.user.routes import namespace_user
from endpoints.v3.permissions.routes import permission_router
from endpoints.v3.source_manager.sync.routes import sm_router
from endpoints.v3.user.routes import user_router
from middleware.scheduled_tasks.check_database_health import check_database_health
from middleware.scheduled_tasks.manager import SchedulerManager
from middleware.util.env import get_env_variable

env = Env()
env.read_env()

NAMESPACES = [
    namespace_callback,
    namespace_request_reset_password,
    namespace_oauth,
    namespace_reset_token_validation,
    namespace_agencies,
    namespace_data_source,
    namespace_login,
    namespace_refresh_session,
    namespace_reset_password,
    namespace_typeahead_suggestions,
    namespace_search,
    namespace_auth,
    namespace_link_to_github,
    namespace_login_with_github,
    namespace_create_test_user,
    namespace_data_requests,
    namespace_url_checker,
    namespace_user,
    namespace_github,
    namespace_notifications,
    namespace_map,
    namespace_signup,
    namespace_match,
    namespace_locations,
    namespace_metrics,
    namespace_admin,
    namespace_contact,
    namespace_metadata,
    namespace_source_collector,
    namespace_validate_email,
    namespace_resend_validation_email,
]


def get_flask_app_cookie_encryption_key() -> str:
    return get_env_variable("FLASK_APP_COOKIE_ENCRYPTION_KEY")


class UpdatedJSONProvider(DefaultJSONProvider):
    def default(self, o):
        if isinstance(o, date) or isinstance(o, datetime):
            return o.isoformat()
        return super().default(o)


def create_flask_app() -> Flask:
    psycopg2_connection = initialize_psycopg_connection()
    config.connection = psycopg2_connection
    api = get_api_with_namespaces()
    app = Flask(__name__)
    app.json = UpdatedJSONProvider(app)

    # JWT settings
    app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET_KEY")
    app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(minutes=15)
    app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=1)

    # Other configuration settings
    app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024

    app.secret_key = get_flask_app_cookie_encryption_key()
    # app.wsgi_app = WSGIMiddleware(app.wsgi_app)
    CORS(app)

    api.init_app(app)
    oauth.init_app(app)
    limiter.init_app(app)
    jwt.init_app(app)

    current_time = datetime.now()

    # Initialize and start the scheduler
    scheduler = SchedulerManager(app)

    flag_run_scheduled_jobs: bool = env.bool("FLAG_RUN_SCHEDULED_JOBS", False)
    if flag_run_scheduled_jobs:
        scheduler.add_job(
            job_id="database_health_check",
            func=check_database_health,
            interval=IntervalTrigger(
                start_date=current_time + timedelta(minutes=3), minutes=60
            ),
        )
        scheduler.add_materialized_view_scheduled_job("typeahead_locations", 1)
        scheduler.add_materialized_view_scheduled_job("typeahead_agencies", 2)
        scheduler.add_materialized_view_scheduled_job("map_states", 4)
        scheduler.add_materialized_view_scheduled_job("map_counties", 5)
        scheduler.add_materialized_view_scheduled_job("map_localities", 6)
        scheduler.start()
    else:
        print("Scheduled jobs are disabled.")
    # Store scheduler in the app context to manage it later
    app.scheduler = scheduler  # pyright: ignore[reportAttributeAccessIssue]

    return app


def get_api_with_namespaces():
    api = Api(
        version="2.0",
        title="PDAP Data Sources API",
        description="The following is the API documentation for the PDAP Data Sources API."
        "\n\nBy accessing our API, you are agreeing to our [Terms of Service](https://docs.pdap.io/meta/operations/legal/terms-of-service). Please read them before you start."
        "\n\nFor API help, consult [our getting started guide.](https://docs.pdap.io/api/introduction)"
        "\n\nTo search the database, go to [pdap.io](https://pdap.io)."
        "\n\nThe new FastAPI API is available at {this_address}/docs",
    )
    for namespace in NAMESPACES:
        api.add_namespace(namespace)
    return api


def create_asgi_app() -> Starlette:
    flask_app = create_flask_app()
    fast_api_app: FastAPI = create_fast_api_app()

    app = Starlette()

    # Add CORS middleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=[
            "https://pdap.io",
            "https://www.pdap.io",
            "https://data-sources.pdap.dev",
            "https://pdap.dev",
            "https://data-sources.pdap.io",
            # Dev origins
            "http://localhost:8888",
        ],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    app.add_middleware(SuppressTestAccessLogMiddleware)

    app.mount("/api/v3", fast_api_app)
    app.mount("/api/v2", WSGIMiddlewareFastAPI(flask_app))

    install_test_request_log_filter()

    return app


def create_fast_api_app() -> FastAPI:
    fast_api_app = FastAPI(
        title="PDAP Data Sources API",
        version="3.0",
        description="The following is the API documentation for the PDAP Data Sources API."
        "\n\nBy accessing our API, you are agreeing to our [Terms of Service](https://docs.pdap.io/meta/operations/legal/terms-of-service). Please read them before you start."
        "\n\nFor API help, consult [our getting started guide.](https://docs.pdap.io/api/introduction)"
        "\n\nTo search the database, go to [pdap.io](https://pdap.io)."
        "\n\nThe old Flask API is available at {this_address}/"
        "",
    )
    for router in [sm_router, user_router, permission_router]:
        fast_api_app.include_router(router)

    return fast_api_app


if __name__ == "__main__":
    app = create_asgi_app()
    uvicorn.run(
        app,
        host=os.getenv("FLASK_RUN_HOST", "127.0.0.1"),
        port=int(os.getenv("FLASK_RUN_PORT", 8000)),
    )