79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
import os
|
||
from flask import Flask, g
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||
from sqlalchemy.orm import sessionmaker
|
||
from motor.motor_asyncio import AsyncIOMotorClient
|
||
from databases.model import Base
|
||
import asyncio
|
||
|
||
# Инициализация Flask
|
||
app = Flask(__name__)
|
||
|
||
# Настройки PostgreSQL из переменных окружения
|
||
POSTGRES_DSN = os.getenv("POSTGRES_URL")
|
||
|
||
# Создание движка для PostgreSQL
|
||
postgres_engine = create_async_engine(POSTGRES_DSN, echo=False)
|
||
AsyncSessionLocal = sessionmaker(bind=postgres_engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
||
# Настройки MongoDB из переменных окружения
|
||
MONGO_URI = os.getenv("MONGO_URL")
|
||
DATABASE_NAME = os.getenv("DB_NAME")
|
||
|
||
# Создание клиента MongoDB
|
||
mongo_client = AsyncIOMotorClient(MONGO_URI)
|
||
mongo_db = mongo_client[DATABASE_NAME]
|
||
|
||
@app.before_first_request
|
||
async def init_databases():
|
||
"""
|
||
Инициализация подключений к PostgreSQL и MongoDB перед первым запросом.
|
||
"""
|
||
try:
|
||
# Инициализация PostgreSQL
|
||
async with postgres_engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
print("PostgreSQL connected.")
|
||
|
||
# Проверка подключения к MongoDB
|
||
await mongo_client.admin.command("ping")
|
||
print("MongoDB connected.")
|
||
except Exception as e:
|
||
print(f"Database initialization failed: {e}")
|
||
|
||
@app.teardown_appcontext
|
||
def close_connections(exception=None):
|
||
"""
|
||
Закрытие соединений с базами данных после окончания работы приложения.
|
||
"""
|
||
asyncio.run(postgres_engine.dispose())
|
||
mongo_client.close()
|
||
print("Database connections closed.")
|
||
|
||
@app.route("/postgres_session")
|
||
async def get_postgres_session():
|
||
"""
|
||
Пример использования сессии PostgreSQL в маршруте Flask.
|
||
"""
|
||
try:
|
||
async with AsyncSessionLocal() as session:
|
||
# Здесь можно выполнить запросы к базе данных PostgreSQL
|
||
result = await session.execute("SELECT 1")
|
||
return {"postgres_result": result.scalar()}
|
||
except Exception as e:
|
||
return {"error": str(e)}, 500
|
||
|
||
@app.route("/mongo_status")
|
||
async def get_mongo_status():
|
||
"""
|
||
Пример проверки MongoDB в маршруте Flask.
|
||
"""
|
||
try:
|
||
await mongo_client.admin.command("ping")
|
||
return {"mongo_status": "connected"}
|
||
except Exception as e:
|
||
return {"error": str(e)}, 500
|
||
|
||
if __name__ == "__main__":
|
||
app.run(debug=True)
|