import datetime, json, os, pytz, random, secrets # , uvicorn from enum import Enum #from datetime import datetime, time, timedelta from fastapi import Body, Cookie, Depends, FastAPI, File, Form, Header, HTTPException, Path, Query, Request, Response, status, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles from functools import lru_cache from pydantic import BaseModel, EmailStr, Field from typing import Dict, List, Optional, Set, Union from . import config # from app.lib_general import common_route_params, Common_Route_Params import logging import app.log # Baby step: Import the new registry module from app.routers.registry import setup_routers # Import middleware with alias to avoid shadowing 'app' FastAPI instance from app.middleware import add_process_time_header as process_time_middleware # Import the routers here first: from app.routers import ae_obj, aether_cfg, api_crud, api_crud_v2, api_crud_v3, api, importing, sql, account, activity_log, address, archive, archive_content, contact, data_store, event, event_abstract, event_badge, event_badge_importing, event_badge_template, event_device, event_exhibit, event_exhibit_tracking, event_file, event_importing, event_location, event_person, event_person_detail, event_person_tracking, event_presentation, event_presenter, event_registration, event_session, flask_cfg, fundraising, grant, hosted_file, log_client_viewing, lookup, membership_cfg, membership_group, membership_person_group, membership_person, membership_person_profile, membership_type, membership_person_type, order, order_v3, order_line, order_cart, organization, page, person, person_user, post, post_comment, product, qr, site, site_domain, user, util_email, websockets_redis, e_confex, e_cvent, e_impexium, e_stripe # from app.routers import agent_bridge # cont_edu_cert, cont_edu_cert_person, # from app.routers import aether_cfg, sql from app.db_sql import sql_select, reset_redis # , sql_connect print('### **** *** ** * The Aether API v4 using FastAPI is loading... * ** *** **** ###') log = logging.getLogger(__name__) # log.setLevel(logging.DEBUG) # DEBUG > INFO > WARNING > ERROR > CRITICAL #logging.basicConfig( #format='[%(asctime)s] %(levelname)s @ %(module)s.%(funcName)s()#%(lineno)d: %(message)s' #) print('### **** *** ** * Aether API v4 using FastAPI - About to try FastAPI() while loading... * ** *** **** ###') app = FastAPI( # debug = True, title = 'Aether API', description = 'One Sky IT\'s Aether API v4 using FastAPI.', version = '4.9.0', operationsSorter = 'method', ) log.setLevel(logging.INFO) # log.debug(config.settings) print('### **** *** ** * Aether API v4 using FastAPI - About to try first SQL SELECT (sql_select()) while loading... * ** *** **** ###') if aether_cfg_sql_result := sql_select( table_name = 'cfg', record_id = config.settings.AETHER_CFG['id'], as_list = False, max_count = 1, ): aether_cfg_sql = aether_cfg_sql_result config.settings.DB['server'] = aether_cfg_sql.get('db_server') config.settings.DB['port'] = aether_cfg_sql.get('db_port') config.settings.DB['name'] = aether_cfg_sql.get('db_name') config.settings.DB['username'] = aether_cfg_sql.get('db_username') config.settings.DB['password'] = aether_cfg_sql.get('db_password') DB = config.settings.DB config.settings.SQLALCHEMY_DB_URI = 'mysql://'+DB['username']+':'+DB['password']+'@'+DB['server']+'/'+DB['name'] # db_result = sql_connect(config.settings.SQLALCHEMY_DB_URI) log.debug(config.settings.DB) config.settings.SMTP['server'] = aether_cfg_sql.get('smtp_server') config.settings.SMTP['port'] = aether_cfg_sql.get('smtp_port') config.settings.SMTP['username'] = aether_cfg_sql.get('smtp_username') config.settings.SMTP['password'] = aether_cfg_sql.get('smtp_password') # config.settings.FILES_PATH['hosted_files_root'] = aether_cfg_sql.get('PATH_HOSTED_FILES_ROOT') # config.settings.FILES_PATH['hosted_tmp_root'] = aether_cfg_sql.get('PATH_HOSTED_TMP_ROOT') config.settings.FILES_PATH['hosted_files_root'] = aether_cfg_sql.get('path_hosted_files_root') config.settings.FILES_PATH['hosted_tmp_root'] = aether_cfg_sql.get('path_hosted_tmp_root') else: # aether_cfg_sql_result pass print('### **** *** ** * Aether API v4 using FastAPI - Finished first SQL SELECT (sql_select()) while loading... * ** *** **** ###' ) log.debug(aether_cfg_sql_result) log.debug(config.settings) # @lru_cache() # def get_settings(): # return config.Settings() app.mount('/static', StaticFiles(directory='static'), name='static') # Register all application routes setup_routers(app) # Legacy registration block being removed... # Set up each route once the router has been imported # app.include_router( # ae_obj.router, # prefix='/ae_obj', # tags=['AE Object'], # ) app.include_router( aether_cfg.router, tags=['Aether Config'], ) # BEGIN: CORS # NOTE: Eventually this should query the DB for the specific list based on the cfg table and or site_domain table. That way it is dynamic and only allowing those defined in the DB. No wildcards or regex. # NOTE: Need to include .localhost for less browser restrictions! Mainly for audio and video. app.add_middleware( CORSMiddleware, # allow_origins = origins, allow_origins = config.settings.ORIGINS, allow_origin_regex = config.settings.ORIGINS_REGEX, # allow_origin_regex = 'https://.*\.oneskyit\.com', allow_credentials = True, allow_methods = ['*'], allow_headers = ['*'], #expose_headers = [], #max_age = 600, ) # END: CORS @app.on_event('startup') async def startup(): log.setLevel(logging.INFO) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) log.info('The Aether FastAPI API is starting up...') #await database.connect() @app.on_event('shutdown') async def shutdown(): log.setLevel(logging.INFO) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) log.info('The Aether FastAPI API is shutting down...') #await database.disconnect() #Add the processing time to the response header. @app.middleware('http') async def add_process_time_header(request: Request, call_next): import time start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers['X-Process-Time'] = str(process_time) return response # ### BEGIN ### API Main ### fastapi_root() ### @app.get('/', tags=['Root'], response_class=PlainTextResponse) async def fastapi_root(response: Response = Response): log.setLevel(logging.DEBUG) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) # log.info(config.settings.APP_NAME) log.info('One Sky IT\'s Aether API root (FastAPI)') log.info('***') log.debug('This is debug') # 10 DEBUG log.info('This is info') # 20 INFO log.warning('This is a warning') # 30 WARNING (and WARN) log.error('This is an error') # 40 ERROR log.exception('This is an exception') # 40 ERROR log.critical('This is critical') # 50 CRITICAL log.info('^^^') log.warning('Resetting Redis...') reset_redis() log.info('Reset Redis') response_data = {} response_data['message'] = 'This is One Sky IT\'s Aether API root (FastAPI).' current_datetime = datetime.datetime.now() current_datetime_string = current_datetime.isoformat() timezone = pytz.timezone("America/New_York") current_datetime_tz = timezone.localize(current_datetime) current_datetime_tz_string = current_datetime_tz.isoformat() current_datetime_utc = datetime.datetime.utcnow() current_datetime_utc_string = current_datetime_utc.isoformat() current_datetime_utc_localize = pytz.utc.localize(current_datetime_utc) current_datetime_utc_localize_string = current_datetime_utc_localize.isoformat() current_datetime_utc_localize_pst = current_datetime_utc_localize.astimezone(pytz.timezone("America/Los_Angeles")) current_datetime_utc_localize_pst_string = current_datetime_utc_localize_pst.isoformat() response_data['datetime'] = current_datetime_string response_data['datetime_tz'] = current_datetime_tz_string response_data['datetime_utc'] = current_datetime_utc_string response_data['datetime_utc_localize'] = current_datetime_utc_localize_string response_data['datetime_utc_localize_pst'] = current_datetime_utc_localize_pst_string response_data['url_safe_string_4_bytes_1'] = secrets.token_urlsafe(4) response_data['url_safe_string_8_bytes_1'] = secrets.token_urlsafe(8) response_data['url_safe_string_8_bytes_2'] = secrets.token_urlsafe(8) response_data['url_safe_string_8_bytes_3'] = secrets.token_urlsafe(8) response_data['url_safe_string_8_bytes_4'] = secrets.token_urlsafe(8) response_data['url_safe_string_8_bytes_5'] = secrets.token_urlsafe(8) response_data['url_safe_string_16_bytes_1'] = secrets.token_urlsafe(16) response_data['url_safe_string_16_bytes_2'] = secrets.token_urlsafe(16) response_data['url_safe_string_16_bytes_3'] = secrets.token_urlsafe(16) response_data['url_safe_string_16_bytes_4'] = secrets.token_urlsafe(16) response_data['url_safe_string_16_bytes_5'] = secrets.token_urlsafe(16) response_data['hex_string_4_bytes_1'] = secrets.token_hex(4) response_data['hex_string_8_bytes_1'] = secrets.token_hex(8) response_data['hex_string_16_bytes_1'] = secrets.token_hex(16) response_data['hex_string_32_bytes_1'] = secrets.token_hex(32) log.debug(json.dumps(response_data, indent=4)) return json.dumps(response_data, indent=4) # , sort_keys=True # ### END ### API Main ### fastapi_root() ### # ### BEGIN ### API Main ### generate_id_random() ### # NOTE: This is just a quick utility function to generate a bunch of random IDs. # Updated 2022-03-30 @app.get('/generate_id_random', tags=['Root'], response_class=PlainTextResponse) async def generate_id_random(response: Response = Response): log.setLevel(logging.INFO) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) response_data = {} html_list = '' return HTMLResponse(content=html_list, status_code=200) # ### END ### API Main ### generate_id_random() ###