merged with dev

services/db.py

@@ -1,21 +1,23 @@
-import json
 import math
 import time
 import traceback
 import warnings
 from typing import Any, Callable, Dict, TypeVar
 
+import orjson
 import sqlalchemy
 from sqlalchemy import (
     JSON,
     Column,
     Engine,
+    Index,
     Integer,
     create_engine,
     event,
     exc,
     func,
     inspect,
+    text,
 )
 from sqlalchemy.orm import Session, configure_mappers, declarative_base
 from sqlalchemy.sql.schema import Table
@@ -56,6 +58,82 @@ def create_table_if_not_exists(engine, table):
         logger.info(f"Table '{table.__tablename__}' ok.")
 
 
+def sync_indexes():
+    """
+    Синхронизирует индексы в БД с индексами, определенными в моделях SQLAlchemy.
+    Создает недостающие индексы, если они определены в моделях, но отсутствуют в БД.
+
+    Использует pg_catalog для PostgreSQL для получения списка существующих индексов.
+    """
+    if not DB_URL.startswith("postgres"):
+        logger.warning("Функция sync_indexes поддерживается только для PostgreSQL.")
+        return
+
+    logger.info("Начинаем синхронизацию индексов в базе данных...")
+
+    # Получаем все существующие индексы в БД
+    with local_session() as session:
+        existing_indexes_query = text("""
+            SELECT 
+                t.relname AS table_name,
+                i.relname AS index_name
+            FROM 
+                pg_catalog.pg_class i
+            JOIN 
+                pg_catalog.pg_index ix ON ix.indexrelid = i.oid
+            JOIN 
+                pg_catalog.pg_class t ON t.oid = ix.indrelid
+            JOIN 
+                pg_catalog.pg_namespace n ON n.oid = i.relnamespace
+            WHERE 
+                i.relkind = 'i'
+                AND n.nspname = 'public'
+                AND t.relkind = 'r'
+            ORDER BY 
+                t.relname, i.relname;
+        """)
+
+        existing_indexes = {row[1].lower() for row in session.execute(existing_indexes_query)}
+        logger.debug(f"Найдено {len(existing_indexes)} существующих индексов в БД")
+
+        # Проверяем каждую модель и её индексы
+        for _model_name, model_class in REGISTRY.items():
+            if hasattr(model_class, "__table__") and hasattr(model_class, "__table_args__"):
+                table_args = model_class.__table_args__
+
+                # Если table_args - это кортеж, ищем в нём объекты Index
+                if isinstance(table_args, tuple):
+                    for arg in table_args:
+                        if isinstance(arg, Index):
+                            index_name = arg.name.lower()
+
+                            # Проверяем, существует ли индекс в БД
+                            if index_name not in existing_indexes:
+                                logger.info(
+                                    f"Создаем отсутствующий индекс {index_name} для таблицы {model_class.__tablename__}"
+                                )
+
+                                # Создаем индекс если он отсутствует
+                                try:
+                                    arg.create(engine)
+                                    logger.info(f"Индекс {index_name} успешно создан")
+                                except Exception as e:
+                                    logger.error(f"Ошибка при создании индекса {index_name}: {e}")
+                            else:
+                                logger.debug(f"Индекс {index_name} уже существует")
+
+        # Анализируем таблицы для оптимизации запросов
+        for model_name, model_class in REGISTRY.items():
+            if hasattr(model_class, "__tablename__"):
+                try:
+                    session.execute(text(f"ANALYZE {model_class.__tablename__}"))
+                    logger.debug(f"Таблица {model_class.__tablename__} проанализирована")
+                except Exception as e:
+                    logger.error(f"Ошибка при анализе таблицы {model_class.__tablename__}: {e}")
+
+    logger.info("Синхронизация индексов завершена.")
+
+
 # noinspection PyUnusedLocal
 def local_session(src=""):
     return Session(bind=engine, expire_on_commit=False)
@@ -84,8 +162,8 @@ class Base(declarative_base()):
                 # Check if the value is JSON and decode it if necessary
                 if isinstance(value, (str, bytes)) and isinstance(self.__table__.columns[column_name].type, JSON):
                     try:
-                        data[column_name] = json.loads(value)
-                    except (TypeError, json.JSONDecodeError) as e:
+                        data[column_name] = orjson.loads(value)
+                    except (TypeError, orjson.JSONDecodeError) as e:
                         logger.error(f"Error decoding JSON for column '{column_name}': {e}")
                         data[column_name] = value
                 else:

services/notify.py

@@ -1,4 +1,4 @@
-import json
+import orjson
 
 from orm.notification import Notification
 from services.db import local_session
@@ -18,7 +18,7 @@ async def notify_reaction(reaction, action: str = "create"):
     data = {"payload": reaction, "action": action}
     try:
         save_notification(action, channel_name, data.get("payload"))
-        await redis.publish(channel_name, json.dumps(data))
+        await redis.publish(channel_name, orjson.dumps(data))
     except Exception as e:
         logger.error(f"Failed to publish to channel {channel_name}: {e}")
 
@@ -28,7 +28,7 @@ async def notify_shout(shout, action: str = "update"):
     data = {"payload": shout, "action": action}
     try:
         save_notification(action, channel_name, data.get("payload"))
-        await redis.publish(channel_name, json.dumps(data))
+        await redis.publish(channel_name, orjson.dumps(data))
     except Exception as e:
         logger.error(f"Failed to publish to channel {channel_name}: {e}")
 
@@ -43,7 +43,7 @@ async def notify_follower(follower: dict, author_id: int, action: str = "follow"
         save_notification(action, channel_name, data.get("payload"))
 
         # Convert data to JSON string
-        json_data = json.dumps(data)
+        json_data = orjson.dumps(data)
 
         # Ensure the data is not empty before publishing
         if json_data:

services/viewed.py

@@ -1,10 +1,11 @@
 import asyncio
-import json
 import os
 import time
 from datetime import datetime, timedelta, timezone
 from typing import Dict
 
+import orjson
+
 # ga
 from google.analytics.data_v1beta import BetaAnalyticsDataClient
 from google.analytics.data_v1beta.types import (
@@ -84,7 +85,7 @@ class ViewedStorage:
                 logger.warn(f" * {viewfile_path} is too old: {self.start_date}")
 
             with open(viewfile_path, "r") as file:
-                precounted_views = json.load(file)
+                precounted_views = orjson.loads(file.read())
                 self.precounted_by_slug.update(precounted_views)
                 logger.info(f" * {len(precounted_views)} shouts with views was loaded.")