22 Commits

Author SHA1 Message Date
fe9984e2d8 type-fix
All checks were successful
Deploy on push / deploy (push) Successful in 43s
2025-03-22 18:39:14 +03:00
369ff757b0 [0.4.16] - 2025-03-22
All checks were successful
Deploy on push / deploy (push) Successful in 6s
- Added hierarchical comments pagination:
  - Created new GraphQL query `load_comments_branch` for efficient loading of hierarchical comments
  - Ability to load root comments with their first N replies
  - Added pagination for both root and child comments
  - Using existing `commented` field in `Stat` type to display number of replies
  - Added special `first_replies` field to store first replies to a comment
  - Optimized SQL queries for efficient loading of comment hierarchies
  - Implemented flexible comment sorting system (by time, rating)
2025-03-22 13:37:43 +03:00
615f1fe468 topics+authors-reimplemented-cache
All checks were successful
Deploy on push / deploy (push) Successful in 5s
2025-03-22 11:47:19 +03:00
86ddb50cb8 topics caching upgrade 2025-03-22 09:31:53 +03:00
31c32143d0 reaction-to-feature-fix
All checks were successful
Deploy on push / deploy (push) Successful in 56s
2025-03-21 12:34:10 +03:00
b63c387806 jsonfix3
All checks were successful
Deploy on push / deploy (push) Successful in 56s
2025-03-20 12:52:44 +03:00
dbbfd42e08 redeploy
All checks were successful
Deploy on push / deploy (push) Successful in 57s
2025-03-20 12:35:55 +03:00
47e12b4452 fx2
Some checks failed
Deploy on push / deploy (push) Failing after 16s
2025-03-20 12:33:27 +03:00
e1a1b4dc7d fx
All checks were successful
Deploy on push / deploy (push) Successful in 44s
2025-03-20 12:25:18 +03:00
ca01181f37 jsonfix
All checks were successful
Deploy on push / deploy (push) Successful in 44s
2025-03-20 12:24:30 +03:00
0aff77eda6 portfix
All checks were successful
Deploy on push / deploy (push) Successful in 49s
2025-03-20 12:13:14 +03:00
8a95aa1209 jsonload-fix
All checks were successful
Deploy on push / deploy (push) Successful in 45s
2025-03-20 12:05:58 +03:00
a4a3c35f4d lesscode
All checks were successful
Deploy on push / deploy (push) Successful in 46s
2025-03-20 12:04:47 +03:00
edece36ecc jsonenc-fix
All checks were successful
Deploy on push / deploy (push) Successful in 12s
2025-03-20 11:59:43 +03:00
247fc98760 cachedep-fix+orjson+fmt
All checks were successful
Deploy on push / deploy (push) Successful in 1m16s
2025-03-20 11:55:21 +03:00
a1781b3800 depfix
All checks were successful
Deploy on push / deploy (push) Successful in 1m4s
2025-03-20 11:36:12 +03:00
450c73c060 nothreads
All checks were successful
Deploy on push / deploy (push) Successful in 47s
2025-03-20 11:30:36 +03:00
3a1924279f redeploy
All checks were successful
Deploy on push / deploy (push) Successful in 49s
2025-03-20 11:23:37 +03:00
094e7e6fe2 granian-fix
Some checks failed
Deploy on push / deploy (push) Failing after 7s
2025-03-20 11:19:29 +03:00
ae48a18536 comment-delete-handling-patch
All checks were successful
Deploy on push / deploy (push) Successful in 1m15s
2025-03-20 11:01:39 +03:00
354bda0efa drafts-fix
All checks were successful
Deploy on push / deploy (push) Successful in 59s
2025-03-13 22:21:43 +03:00
856f4ffc85 i 2025-03-09 21:01:52 +03:00
35 changed files with 2170 additions and 1014 deletions

View File

@@ -29,16 +29,7 @@ jobs:
if: github.ref == 'refs/heads/dev'
uses: dokku/github-action@master
with:
branch: 'main'
branch: 'dev'
force: true
git_remote_url: 'ssh://dokku@v2.discours.io:22/core'
ssh_private_key: ${{ secrets.SSH_PRIVATE_KEY }}
- name: Push to dokku for staging branch
if: github.ref == 'refs/heads/staging'
uses: dokku/github-action@master
with:
branch: 'dev'
git_remote_url: 'ssh://dokku@staging.discours.io:22/core'
ssh_private_key: ${{ secrets.SSH_PRIVATE_KEY }}
git_push_flags: '--force'

3
.gitignore vendored
View File

@@ -161,5 +161,4 @@ views.json
*.key
*.crt
*cache.json
.cursor
.devcontainer/
.cursor

View File

@@ -1,3 +1,83 @@
#### [0.4.16] - 2025-03-22
- Added hierarchical comments pagination:
- Created new GraphQL query `load_comments_branch` for efficient loading of hierarchical comments
- Ability to load root comments with their first N replies
- Added pagination for both root and child comments
- Using existing `commented` field in `Stat` type to display number of replies
- Added special `first_replies` field to store first replies to a comment
- Optimized SQL queries for efficient loading of comment hierarchies
- Implemented flexible comment sorting system (by time, rating)
#### [0.4.15] - 2025-03-22
- Upgraded caching system described `docs/caching.md`
- Module `cache/memorycache.py` removed
- Enhanced caching system with backward compatibility:
- Unified cache key generation with support for existing naming patterns
- Improved Redis operation function with better error handling
- Updated precache module to use consistent Redis interface
- Integrated revalidator with the invalidation system for better performance
- Added comprehensive documentation for the caching system
- Enhanced cached_query to support template-based cache keys
- Standardized error handling across all cache operations
- Optimized cache invalidation system:
- Added targeted invalidation for individual entities (authors, topics)
- Improved revalidation manager with individual object processing
- Implemented batched processing for high-volume invalidations
- Reduced Redis operations by using precise key invalidation instead of prefix-based wipes
- Added special handling for slug changes in topics
- Unified caching system for all models:
- Implemented abstract functions `cache_data`, `get_cached_data` and `invalidate_cache_by_prefix`
- Added `cached_query` function for unified approach to query caching
- Updated resolvers `author.py` and `topic.py` to use the new caching API
- Improved logging for cache operations to simplify debugging
- Optimized Redis memory usage through key format unification
- Improved caching and sorting in Topic and Author modules:
- Added support for dictionary sorting parameters in `by` for both modules
- Optimized cache key generation for stable behavior with various parameters
- Enhanced sorting logic with direction support and arbitrary fields
- Added `by` parameter support in the API for getting topics by community
- Performance optimizations for author-related queries:
- Added SQLAlchemy-managed indexes to `Author`, `AuthorFollower`, `AuthorRating` and `AuthorBookmark` models
- Implemented persistent Redis caching for author queries without TTL (invalidated only on changes)
- Optimized author retrieval with separate endpoints:
- `get_authors_all` - returns all non-deleted authors without statistics
- `get_authors_paginated` - returns authors with statistics and pagination support
- `load_authors_by` - optimized to use caching and efficient sorting
- Improved SQL queries with optimized JOIN conditions and efficient filtering
- Added pre-aggregation of statistics (shouts count, followers count) in single efficient queries
- Implemented robust cache invalidation on author updates
- Created necessary indexes for author lookups by user ID, slug, and timestamps
#### [0.4.14] - 2025-03-21
- Significant performance improvements for topic queries:
- Added database indexes to optimize JOIN operations
- Implemented persistent Redis caching for topic queries (no TTL, invalidated only on changes)
- Optimized topic retrieval with separate endpoints for different use cases:
- `get_topics_all` - returns all topics without statistics for lightweight listing
- `get_topics_paginated` - returns topics with statistics and pagination support
- `get_topics_by_community` - adds pagination and optimized filtering by community
- Added SQLAlchemy-managed indexes directly in ORM models for automatic schema maintenance
- Created `sync_indexes()` function for automatic index synchronization during app startup
- Reduced database load by pre-aggregating statistics in optimized SQL queries
- Added robust cache invalidation on topic create/update/delete operations
- Improved query optimization with proper JOIN conditions and specific partial indexes
#### [0.4.13] - 2025-03-20
- Fixed Topic objects serialization error in cache/memorycache.py
- Improved CustomJSONEncoder to support SQLAlchemy models with dict() method
- Enhanced error handling in cache_on_arguments decorator
- Modified `load_reactions_by` to include deleted reactions when `include_deleted=true` for proper comment tree building
- Fixed featured/unfeatured logic in reaction processing:
- Dislike reactions now properly take precedence over likes
- Featured status now requires more than 4 likes from users with featured articles
- Removed unnecessary filters for deleted reactions since rating reactions are physically deleted
- Author's featured status now based on having non-deleted articles with featured_at
#### [0.4.12] - 2025-03-19
- `delete_reaction` detects comments and uses `deleted_at` update
- `check_to_unfeature` etc. update
- dogpile dep in `services/memorycache.py` optimized
#### [0.4.11] - 2025-02-12
- `create_draft` resolver requires draft_id fixed
- `create_draft` resolver defaults body and title fields to empty string
@@ -228,22 +308,4 @@
#### [0.2.7]
- `loadFollowedReactions` now with `login_required`
- notifier service api draft
- added `shout` visibility kind in schema
- community isolated from author in orm
#### [0.2.6]
- redis connection pool
- auth context fixes
- communities orm, resolvers, schema
#### [0.2.5]
- restructured
- all users have their profiles as authors in core
- `gittask`, `inbox` and `auth` logics removed
- `settings` moved to base and now smaller
- new outside auth schema
- removed `gittask`, `auth`, `inbox`, `migration`
- `loadFollowedReactions` now with `

View File

@@ -13,4 +13,6 @@ RUN pip install -r requirements.txt
COPY . .
CMD ["python", "server.py"]
EXPOSE 8000
CMD ["python", "-m", "granian", "main:app", "--interface", "asgi", "--host", "0.0.0.0", "--port", "8000"]

271
cache/cache.py vendored
View File

@@ -1,7 +1,37 @@
"""
Caching system for the Discours platform
----------------------------------------
This module provides a comprehensive caching solution with these key components:
1. KEY NAMING CONVENTIONS:
- Entity-based keys: "entity:property:value" (e.g., "author:id:123")
- Collection keys: "entity:collection:params" (e.g., "authors:stats:limit=10:offset=0")
- Special case keys: Maintained for backwards compatibility (e.g., "topic_shouts_123")
2. CORE FUNCTIONS:
- cached_query(): High-level function for retrieving cached data or executing queries
3. ENTITY-SPECIFIC FUNCTIONS:
- cache_author(), cache_topic(): Cache entity data
- get_cached_author(), get_cached_topic(): Retrieve entity data from cache
- invalidate_cache_by_prefix(): Invalidate all keys with a specific prefix
4. CACHE INVALIDATION STRATEGY:
- Direct invalidation via invalidate_* functions for immediate changes
- Delayed invalidation via revalidation_manager for background processing
- Event-based triggers for automatic cache updates (see triggers.py)
To maintain consistency with the existing codebase, this module preserves
the original key naming patterns while providing a more structured approach
for new cache operations.
"""
import asyncio
import json
from typing import List
from typing import Any, Dict, List, Optional, Union
import orjson
from sqlalchemy import and_, join, select
from orm.author import Author, AuthorFollower
@@ -19,8 +49,10 @@ DEFAULT_FOLLOWS = {
"communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
}
CACHE_TTL = 300 # 5 минут
CACHE_TTL = 300 # 5 minutes
# Key templates for common entity types
# These are used throughout the codebase and should be maintained for compatibility
CACHE_KEYS = {
"TOPIC_ID": "topic:id:{}",
"TOPIC_SLUG": "topic:slug:{}",
@@ -37,8 +69,8 @@ CACHE_KEYS = {
async def cache_topic(topic: dict):
payload = json.dumps(topic, cls=CustomJSONEncoder)
await asyncio.gather(
redis_operation("SET", f"topic:id:{topic['id']}", payload),
redis_operation("SET", f"topic:slug:{topic['slug']}", payload),
redis.execute("SET", f"topic:id:{topic['id']}", payload),
redis.execute("SET", f"topic:slug:{topic['slug']}", payload),
)
@@ -46,30 +78,30 @@ async def cache_topic(topic: dict):
async def cache_author(author: dict):
payload = json.dumps(author, cls=CustomJSONEncoder)
await asyncio.gather(
redis_operation("SET", f"author:user:{author['user'].strip()}", str(author["id"])),
redis_operation("SET", f"author:id:{author['id']}", payload),
redis.execute("SET", f"author:user:{author['user'].strip()}", str(author["id"])),
redis.execute("SET", f"author:id:{author['id']}", payload),
)
# Cache follows data
async def cache_follows(follower_id: int, entity_type: str, entity_id: int, is_insert=True):
key = f"author:follows-{entity_type}s:{follower_id}"
follows_str = await redis_operation("GET", key)
follows = json.loads(follows_str) if follows_str else DEFAULT_FOLLOWS[entity_type]
follows_str = await redis.execute("GET", key)
follows = orjson.loads(follows_str) if follows_str else DEFAULT_FOLLOWS[entity_type]
if is_insert:
if entity_id not in follows:
follows.append(entity_id)
else:
follows = [eid for eid in follows if eid != entity_id]
await redis_operation("SET", key, json.dumps(follows, cls=CustomJSONEncoder))
await redis.execute("SET", key, json.dumps(follows, cls=CustomJSONEncoder))
await update_follower_stat(follower_id, entity_type, len(follows))
# Update follower statistics
async def update_follower_stat(follower_id, entity_type, count):
follower_key = f"author:id:{follower_id}"
follower_str = await redis_operation("GET", follower_key)
follower = json.loads(follower_str) if follower_str else None
follower_str = await redis.execute("GET", follower_key)
follower = orjson.loads(follower_str) if follower_str else None
if follower:
follower["stat"] = {f"{entity_type}s": count}
await cache_author(follower)
@@ -78,9 +110,9 @@ async def update_follower_stat(follower_id, entity_type, count):
# Get author from cache
async def get_cached_author(author_id: int, get_with_stat):
author_key = f"author:id:{author_id}"
result = await redis_operation("GET", author_key)
result = await redis.execute("GET", author_key)
if result:
return json.loads(result)
return orjson.loads(result)
# Load from database if not found in cache
q = select(Author).where(Author.id == author_id)
authors = get_with_stat(q)
@@ -103,16 +135,16 @@ async def get_cached_topic(topic_id: int):
dict: Topic data or None if not found.
"""
topic_key = f"topic:id:{topic_id}"
cached_topic = await redis_operation("GET", topic_key)
cached_topic = await redis.execute("GET", topic_key)
if cached_topic:
return json.loads(cached_topic)
return orjson.loads(cached_topic)
# If not in cache, fetch from the database
with local_session() as session:
topic = session.execute(select(Topic).where(Topic.id == topic_id)).scalar_one_or_none()
if topic:
topic_dict = topic.dict()
await redis_operation("SET", topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
await redis.execute("SET", topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
return topic_dict
return None
@@ -121,9 +153,9 @@ async def get_cached_topic(topic_id: int):
# Get topic by slug from cache
async def get_cached_topic_by_slug(slug: str, get_with_stat):
topic_key = f"topic:slug:{slug}"
result = await redis_operation("GET", topic_key)
result = await redis.execute("GET", topic_key)
if result:
return json.loads(result)
return orjson.loads(result)
# Load from database if not found in cache
topic_query = select(Topic).where(Topic.slug == slug)
topics = get_with_stat(topic_query)
@@ -138,8 +170,8 @@ async def get_cached_topic_by_slug(slug: str, get_with_stat):
async def get_cached_authors_by_ids(author_ids: List[int]) -> List[dict]:
# Fetch all author data concurrently
keys = [f"author:id:{author_id}" for author_id in author_ids]
results = await asyncio.gather(*(redis_operation("GET", key) for key in keys))
authors = [json.loads(result) if result else None for result in results]
results = await asyncio.gather(*(redis.execute("GET", key) for key in keys))
authors = [orjson.loads(result) if result else None for result in results]
# Load missing authors from database and cache
missing_indices = [index for index, author in enumerate(authors) if author is None]
if missing_indices:
@@ -165,10 +197,10 @@ async def get_cached_topic_followers(topic_id: int):
"""
try:
cache_key = CACHE_KEYS["TOPIC_FOLLOWERS"].format(topic_id)
cached = await redis_operation("GET", cache_key)
cached = await redis.execute("GET", cache_key)
if cached:
followers_ids = json.loads(cached)
followers_ids = orjson.loads(cached)
logger.debug(f"Found {len(followers_ids)} cached followers for topic #{topic_id}")
return await get_cached_authors_by_ids(followers_ids)
@@ -181,7 +213,7 @@ async def get_cached_topic_followers(topic_id: int):
.all()
]
await redis_operation("SETEX", cache_key, value=json.dumps(followers_ids), ttl=CACHE_TTL)
await redis.execute("SETEX", cache_key, CACHE_TTL, orjson.dumps(followers_ids))
followers = await get_cached_authors_by_ids(followers_ids)
logger.debug(f"Cached {len(followers)} followers for topic #{topic_id}")
return followers
@@ -194,9 +226,9 @@ async def get_cached_topic_followers(topic_id: int):
# Get cached author followers
async def get_cached_author_followers(author_id: int):
# Check cache for data
cached = await redis_operation("GET", f"author:followers:{author_id}")
cached = await redis.execute("GET", f"author:followers:{author_id}")
if cached:
followers_ids = json.loads(cached)
followers_ids = orjson.loads(cached)
followers = await get_cached_authors_by_ids(followers_ids)
logger.debug(f"Cached followers for author #{author_id}: {len(followers)}")
return followers
@@ -210,7 +242,7 @@ async def get_cached_author_followers(author_id: int):
.filter(AuthorFollower.author == author_id, Author.id != author_id)
.all()
]
await redis_operation("SET", f"author:followers:{author_id}", json.dumps(followers_ids))
await redis.execute("SET", f"author:followers:{author_id}", orjson.dumps(followers_ids))
followers = await get_cached_authors_by_ids(followers_ids)
return followers
@@ -218,9 +250,9 @@ async def get_cached_author_followers(author_id: int):
# Get cached follower authors
async def get_cached_follower_authors(author_id: int):
# Attempt to retrieve authors from cache
cached = await redis_operation("GET", f"author:follows-authors:{author_id}")
cached = await redis.execute("GET", f"author:follows-authors:{author_id}")
if cached:
authors_ids = json.loads(cached)
authors_ids = orjson.loads(cached)
else:
# Query authors from database
with local_session() as session:
@@ -232,7 +264,7 @@ async def get_cached_follower_authors(author_id: int):
.where(AuthorFollower.follower == author_id)
).all()
]
await redis_operation("SET", f"author:follows-authors:{author_id}", json.dumps(authors_ids))
await redis.execute("SET", f"author:follows-authors:{author_id}", orjson.dumps(authors_ids))
authors = await get_cached_authors_by_ids(authors_ids)
return authors
@@ -241,9 +273,9 @@ async def get_cached_follower_authors(author_id: int):
# Get cached follower topics
async def get_cached_follower_topics(author_id: int):
# Attempt to retrieve topics from cache
cached = await redis_operation("GET", f"author:follows-topics:{author_id}")
cached = await redis.execute("GET", f"author:follows-topics:{author_id}")
if cached:
topics_ids = json.loads(cached)
topics_ids = orjson.loads(cached)
else:
# Load topics from database and cache them
with local_session() as session:
@@ -254,13 +286,13 @@ async def get_cached_follower_topics(author_id: int):
.where(TopicFollower.follower == author_id)
.all()
]
await redis_operation("SET", f"author:follows-topics:{author_id}", json.dumps(topics_ids))
await redis.execute("SET", f"author:follows-topics:{author_id}", orjson.dumps(topics_ids))
topics = []
for topic_id in topics_ids:
topic_str = await redis_operation("GET", f"topic:id:{topic_id}")
topic_str = await redis.execute("GET", f"topic:id:{topic_id}")
if topic_str:
topic = json.loads(topic_str)
topic = orjson.loads(topic_str)
if topic and topic not in topics:
topics.append(topic)
@@ -280,12 +312,12 @@ async def get_cached_author_by_user_id(user_id: str, get_with_stat):
dict: Dictionary with author data or None if not found.
"""
# Attempt to find author ID by user_id in Redis cache
author_id = await redis_operation("GET", f"author:user:{user_id.strip()}")
author_id = await redis.execute("GET", f"author:user:{user_id.strip()}")
if author_id:
# If ID is found, get full author data by ID
author_data = await redis_operation("GET", f"author:id:{author_id}")
author_data = await redis.execute("GET", f"author:id:{author_id}")
if author_data:
return json.loads(author_data)
return orjson.loads(author_data)
# If data is not found in cache, query the database
author_query = select(Author).where(Author.user == user_id)
@@ -295,8 +327,8 @@ async def get_cached_author_by_user_id(user_id: str, get_with_stat):
author = authors[0]
author_dict = author.dict()
await asyncio.gather(
redis_operation("SET", f"author:user:{user_id.strip()}", str(author.id)),
redis_operation("SET", f"author:id:{author.id}", json.dumps(author_dict)),
redis.execute("SET", f"author:user:{user_id.strip()}", str(author.id)),
redis.execute("SET", f"author:id:{author.id}", orjson.dumps(author_dict)),
)
return author_dict
@@ -317,9 +349,9 @@ async def get_cached_topic_authors(topic_id: int):
"""
# Attempt to get a list of author IDs from cache
rkey = f"topic:authors:{topic_id}"
cached_authors_ids = await redis_operation("GET", rkey)
cached_authors_ids = await redis.execute("GET", rkey)
if cached_authors_ids:
authors_ids = json.loads(cached_authors_ids)
authors_ids = orjson.loads(cached_authors_ids)
else:
# If cache is empty, get data from the database
with local_session() as session:
@@ -331,7 +363,7 @@ async def get_cached_topic_authors(topic_id: int):
)
authors_ids = [author_id for (author_id,) in session.execute(query).all()]
# Cache the retrieved author IDs
await redis_operation("SET", rkey, json.dumps(authors_ids))
await redis.execute("SET", rkey, orjson.dumps(authors_ids))
# Retrieve full author details from cached IDs
if authors_ids:
@@ -352,11 +384,11 @@ async def invalidate_shouts_cache(cache_keys: List[str]):
cache_key = f"shouts:{key}"
# Удаляем основной кэш
await redis_operation("DEL", cache_key)
await redis.execute("DEL", cache_key)
logger.debug(f"Invalidated cache key: {cache_key}")
# Добавляем ключ в список инвалидированных с TTL
await redis_operation("SETEX", f"{cache_key}:invalidated", value="1", ttl=CACHE_TTL)
await redis.execute("SETEX", f"{cache_key}:invalidated", CACHE_TTL, "1")
# Если это кэш темы, инвалидируем также связанные ключи
if key.startswith("topic_"):
@@ -368,7 +400,7 @@ async def invalidate_shouts_cache(cache_keys: List[str]):
f"topic:stats:{topic_id}",
]
for related_key in related_keys:
await redis_operation("DEL", related_key)
await redis.execute("DEL", related_key)
logger.debug(f"Invalidated related key: {related_key}")
except Exception as e:
@@ -379,15 +411,15 @@ async def cache_topic_shouts(topic_id: int, shouts: List[dict]):
"""Кэширует список публикаций для темы"""
key = f"topic_shouts_{topic_id}"
payload = json.dumps(shouts, cls=CustomJSONEncoder)
await redis_operation("SETEX", key, value=payload, ttl=CACHE_TTL)
await redis.execute("SETEX", key, CACHE_TTL, payload)
async def get_cached_topic_shouts(topic_id: int) -> List[dict]:
"""Получает кэшированный список публикаций для темы"""
key = f"topic_shouts_{topic_id}"
cached = await redis_operation("GET", key)
cached = await redis.execute("GET", key)
if cached:
return json.loads(cached)
return orjson.loads(cached)
return None
@@ -431,27 +463,7 @@ async def invalidate_shout_related_cache(shout: Shout, author_id: int):
await invalidate_shouts_cache(list(cache_keys))
async def redis_operation(operation: str, key: str, value=None, ttl=None):
"""
Унифицированная функция для работы с Redis
Args:
operation: 'GET', 'SET', 'DEL', 'SETEX'
key: ключ
value: значение (для SET/SETEX)
ttl: время жизни в секундах (для SETEX)
"""
try:
if operation == "GET":
return await redis.execute("GET", key)
elif operation == "SET":
await redis.execute("SET", key, value)
elif operation == "SETEX":
await redis.execute("SETEX", key, ttl or CACHE_TTL, value)
elif operation == "DEL":
await redis.execute("DEL", key)
except Exception as e:
logger.error(f"Redis {operation} error for key {key}: {e}")
# Function removed - direct Redis calls used throughout the module instead
async def get_cached_entity(entity_type: str, entity_id: int, get_method, cache_method):
@@ -465,9 +477,9 @@ async def get_cached_entity(entity_type: str, entity_id: int, get_method, cache_
cache_method: метод кэширования
"""
key = f"{entity_type}:id:{entity_id}"
cached = await redis_operation("GET", key)
cached = await redis.execute("GET", key)
if cached:
return json.loads(cached)
return orjson.loads(cached)
entity = await get_method(entity_id)
if entity:
@@ -496,3 +508,120 @@ async def cache_by_id(entity, entity_id: int, cache_method):
d = x.dict()
await cache_method(d)
return d
# Универсальная функция для сохранения данных в кеш
async def cache_data(key: str, data: Any, ttl: Optional[int] = None) -> None:
"""
Сохраняет данные в кеш по указанному ключу.
Args:
key: Ключ кеша
data: Данные для сохранения
ttl: Время жизни кеша в секундах (None - бессрочно)
"""
try:
payload = json.dumps(data, cls=CustomJSONEncoder)
if ttl:
await redis.execute("SETEX", key, ttl, payload)
else:
await redis.execute("SET", key, payload)
logger.debug(f"Данные сохранены в кеш по ключу {key}")
except Exception as e:
logger.error(f"Ошибка при сохранении данных в кеш: {e}")
# Универсальная функция для получения данных из кеша
async def get_cached_data(key: str) -> Optional[Any]:
"""
Получает данные из кеша по указанному ключу.
Args:
key: Ключ кеша
Returns:
Any: Данные из кеша или None, если данных нет
"""
try:
cached_data = await redis.execute("GET", key)
if cached_data:
logger.debug(f"Данные получены из кеша по ключу {key}")
return orjson.loads(cached_data)
return None
except Exception as e:
logger.error(f"Ошибка при получении данных из кеша: {e}")
return None
# Универсальная функция для инвалидации кеша по префиксу
async def invalidate_cache_by_prefix(prefix: str) -> None:
"""
Инвалидирует все ключи кеша с указанным префиксом.
Args:
prefix: Префикс ключей кеша для инвалидации
"""
try:
keys = await redis.execute("KEYS", f"{prefix}:*")
if keys:
await redis.execute("DEL", *keys)
logger.debug(f"Удалено {len(keys)} ключей кеша с префиксом {prefix}")
except Exception as e:
logger.error(f"Ошибка при инвалидации кеша: {e}")
# Универсальная функция для получения и кеширования данных
async def cached_query(
cache_key: str,
query_func: callable,
ttl: Optional[int] = None,
force_refresh: bool = False,
use_key_format: bool = True,
**query_params,
) -> Any:
"""
Gets data from cache or executes query and saves result to cache.
Supports existing key formats for compatibility.
Args:
cache_key: Cache key or key template from CACHE_KEYS
query_func: Function to execute the query
ttl: Cache TTL in seconds (None - indefinite)
force_refresh: Force cache refresh
use_key_format: Whether to check if cache_key matches a key template in CACHE_KEYS
**query_params: Parameters to pass to the query function
Returns:
Any: Data from cache or query result
"""
# Check if cache_key matches a pattern in CACHE_KEYS
actual_key = cache_key
if use_key_format and "{}" in cache_key:
# Look for a template match in CACHE_KEYS
for key_name, key_format in CACHE_KEYS.items():
if cache_key == key_format:
# We have a match, now look for the id or value to format with
for param_name, param_value in query_params.items():
if param_name in ["id", "slug", "user", "topic_id", "author_id"]:
actual_key = cache_key.format(param_value)
break
# If not forcing refresh, try to get data from cache
if not force_refresh:
cached_result = await get_cached_data(actual_key)
if cached_result is not None:
return cached_result
# If data not in cache or refresh required, execute query
try:
result = await query_func(**query_params)
if result is not None:
# Save result to cache
await cache_data(actual_key, result, ttl)
return result
except Exception as e:
logger.error(f"Error executing query for caching: {e}")
# In case of error, return data from cache if not forcing refresh
if not force_refresh:
return await get_cached_data(actual_key)
raise

11
cache/memorycache.py vendored
View File

@@ -1,11 +0,0 @@
from dogpile.cache import make_region
from settings import REDIS_URL
# Создание региона кэша с TTL
cache_region = make_region()
cache_region.configure(
"dogpile.cache.redis",
arguments={"url": f"{REDIS_URL}/1"},
expiration_time=3600, # Cache expiration time in seconds
)

14
cache/precache.py vendored
View File

@@ -86,11 +86,15 @@ async def precache_data():
# Преобразуем словарь в список аргументов для HSET
if value:
flattened = []
for field, val in value.items():
flattened.extend([field, val])
await redis.execute("HSET", key, *flattened)
# Если значение - словарь, преобразуем его в плоский список для HSET
if isinstance(value, dict):
flattened = []
for field, val in value.items():
flattened.extend([field, val])
await redis.execute("HSET", key, *flattened)
else:
# Предполагаем, что значение уже содержит список
await redis.execute("HSET", key, *value)
logger.info(f"redis hash '{key}' was restored")
with local_session() as session:

122
cache/revalidator.py vendored
View File

@@ -1,17 +1,26 @@
import asyncio
from cache.cache import cache_author, cache_topic, get_cached_author, get_cached_topic
from cache.cache import (
cache_author,
cache_topic,
get_cached_author,
get_cached_topic,
invalidate_cache_by_prefix,
)
from resolvers.stat import get_with_stat
from utils.logger import root_logger as logger
CACHE_REVALIDATION_INTERVAL = 300 # 5 minutes
class CacheRevalidationManager:
def __init__(self, interval=60):
def __init__(self, interval=CACHE_REVALIDATION_INTERVAL):
"""Инициализация менеджера с заданным интервалом проверки (в секундах)."""
self.interval = interval
self.items_to_revalidate = {"authors": set(), "topics": set(), "shouts": set(), "reactions": set()}
self.lock = asyncio.Lock()
self.running = True
self.MAX_BATCH_SIZE = 10 # Максимальное количество элементов для поштучной обработки
async def start(self):
"""Запуск фонового воркера для ревалидации кэша."""
@@ -32,22 +41,107 @@ class CacheRevalidationManager:
"""Обновление кэша для всех сущностей, требующих ревалидации."""
async with self.lock:
# Ревалидация кэша авторов
for author_id in self.items_to_revalidate["authors"]:
author = await get_cached_author(author_id, get_with_stat)
if author:
await cache_author(author)
self.items_to_revalidate["authors"].clear()
if self.items_to_revalidate["authors"]:
logger.debug(f"Revalidating {len(self.items_to_revalidate['authors'])} authors")
for author_id in self.items_to_revalidate["authors"]:
if author_id == "all":
await invalidate_cache_by_prefix("authors")
break
author = await get_cached_author(author_id, get_with_stat)
if author:
await cache_author(author)
self.items_to_revalidate["authors"].clear()
# Ревалидация кэша тем
for topic_id in self.items_to_revalidate["topics"]:
topic = await get_cached_topic(topic_id)
if topic:
await cache_topic(topic)
self.items_to_revalidate["topics"].clear()
if self.items_to_revalidate["topics"]:
logger.debug(f"Revalidating {len(self.items_to_revalidate['topics'])} topics")
for topic_id in self.items_to_revalidate["topics"]:
if topic_id == "all":
await invalidate_cache_by_prefix("topics")
break
topic = await get_cached_topic(topic_id)
if topic:
await cache_topic(topic)
self.items_to_revalidate["topics"].clear()
# Ревалидация шаутов (публикаций)
if self.items_to_revalidate["shouts"]:
shouts_count = len(self.items_to_revalidate["shouts"])
logger.debug(f"Revalidating {shouts_count} shouts")
# Проверяем наличие специального флага 'all'
if "all" in self.items_to_revalidate["shouts"]:
await invalidate_cache_by_prefix("shouts")
# Если элементов много, но не 'all', используем специфический подход
elif shouts_count > self.MAX_BATCH_SIZE:
# Инвалидируем только collections keys, которые затрагивают много сущностей
collection_keys = await asyncio.create_task(self._redis.execute("KEYS", "shouts:*"))
if collection_keys:
await self._redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей шаутов")
# Обновляем кеш каждого конкретного шаута
for shout_id in self.items_to_revalidate["shouts"]:
if shout_id != "all":
# Точечная инвалидация для каждого shout_id
specific_keys = [f"shout:id:{shout_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
else:
# Если элементов немного, обрабатываем каждый
for shout_id in self.items_to_revalidate["shouts"]:
if shout_id != "all":
# Точечная инвалидация для каждого shout_id
specific_keys = [f"shout:id:{shout_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
self.items_to_revalidate["shouts"].clear()
# Аналогично для реакций - точечная инвалидация
if self.items_to_revalidate["reactions"]:
reactions_count = len(self.items_to_revalidate["reactions"])
logger.debug(f"Revalidating {reactions_count} reactions")
if "all" in self.items_to_revalidate["reactions"]:
await invalidate_cache_by_prefix("reactions")
elif reactions_count > self.MAX_BATCH_SIZE:
# Инвалидируем только collections keys для реакций
collection_keys = await asyncio.create_task(self._redis.execute("KEYS", "reactions:*"))
if collection_keys:
await self._redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей реакций")
# Точечная инвалидация для каждой реакции
for reaction_id in self.items_to_revalidate["reactions"]:
if reaction_id != "all":
specific_keys = [f"reaction:id:{reaction_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
else:
# Точечная инвалидация для каждой реакции
for reaction_id in self.items_to_revalidate["reactions"]:
if reaction_id != "all":
specific_keys = [f"reaction:id:{reaction_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
self.items_to_revalidate["reactions"].clear()
def mark_for_revalidation(self, entity_id, entity_type):
"""Отметить сущность для ревалидации."""
self.items_to_revalidate[entity_type].add(entity_id)
if entity_id and entity_type:
self.items_to_revalidate[entity_type].add(entity_id)
def invalidate_all(self, entity_type):
"""Пометить для инвалидации все элементы указанного типа."""
logger.debug(f"Marking all {entity_type} for invalidation")
# Особый флаг для полной инвалидации
self.items_to_revalidate[entity_type].add("all")
async def stop(self):
"""Остановка фонового воркера."""
@@ -60,4 +154,4 @@ class CacheRevalidationManager:
pass
revalidation_manager = CacheRevalidationManager(interval=300) # Ревалидация каждые 5 минут
revalidation_manager = CacheRevalidationManager()

279
docs/caching.md Normal file
View File

@@ -0,0 +1,279 @@
# Система кеширования Discours
## Общее описание
Система кеширования Discours - это комплексное решение для повышения производительности платформы. Она использует Redis для хранения часто запрашиваемых данных и уменьшения нагрузки на основную базу данных.
Кеширование реализовано как многоуровневая система, состоящая из нескольких модулей:
- `cache.py` - основной модуль с функциями кеширования
- `revalidator.py` - асинхронный менеджер ревалидации кеша
- `triggers.py` - триггеры событий SQLAlchemy для автоматической ревалидации
- `precache.py` - предварительное кеширование данных при старте приложения
## Ключевые компоненты
### 1. Форматы ключей кеша
Система поддерживает несколько форматов ключей для обеспечения совместимости и удобства использования:
- **Ключи сущностей**: `entity:property:value` (например, `author:id:123`)
- **Ключи коллекций**: `entity:collection:params` (например, `authors:stats:limit=10:offset=0`)
- **Специальные ключи**: для обратной совместимости (например, `topic_shouts_123`)
Все стандартные форматы ключей хранятся в словаре `CACHE_KEYS`:
```python
CACHE_KEYS = {
"TOPIC_ID": "topic:id:{}",
"TOPIC_SLUG": "topic:slug:{}",
"AUTHOR_ID": "author:id:{}",
# и другие...
}
```
### 2. Основные функции кеширования
#### Структура ключей
Вместо генерации ключей через вспомогательные функции, система следует строгим конвенциям формирования ключей:
1. **Ключи для отдельных сущностей** строятся по шаблону:
```
entity:property:value
```
Например:
- `topic:id:123` - тема с ID 123
- `author:slug:john-doe` - автор со слагом "john-doe"
- `shout:id:456` - публикация с ID 456
2. **Ключи для коллекций** строятся по шаблону:
```
entity:collection[:filter1=value1:filter2=value2:...]
```
Например:
- `topics:all:basic` - базовый список всех тем
- `authors:stats:limit=10:offset=0:sort=name` - отсортированный список авторов с пагинацией
- `shouts:feed:limit=20:community=1` - лента публикаций с фильтром по сообществу
3. **Специальные форматы ключей** для обратной совместимости:
```
entity_action_id
```
Например:
- `topic_shouts_123` - публикации для темы с ID 123
Во всех модулях системы разработчики должны явно формировать ключи в соответствии с этими конвенциями, что обеспечивает единообразие и предсказуемость кеширования.
#### Работа с данными в кеше
```python
async def cache_data(key, data, ttl=None)
async def get_cached_data(key)
```
Эти функции предоставляют универсальный интерфейс для сохранения и получения данных из кеша. Они напрямую используют Redis через вызовы `redis.execute()`.
#### Высокоуровневое кеширование запросов
```python
async def cached_query(cache_key, query_func, ttl=None, force_refresh=False, **query_params)
```
Функция `cached_query` объединяет получение данных из кеша и выполнение запроса в случае отсутствия данных в кеше. Это основная функция, которую следует использовать в резолверах для кеширования результатов запросов.
### 3. Кеширование сущностей
Для основных типов сущностей реализованы специальные функции:
```python
async def cache_topic(topic: dict)
async def cache_author(author: dict)
async def get_cached_topic(topic_id: int)
async def get_cached_author(author_id: int, get_with_stat)
```
Эти функции упрощают работу с часто используемыми типами данных и обеспечивают единообразный подход к их кешированию.
### 4. Работа со связями
Для работы со связями между сущностями предназначены функции:
```python
async def cache_follows(follower_id, entity_type, entity_id, is_insert=True)
async def get_cached_topic_followers(topic_id)
async def get_cached_author_followers(author_id)
async def get_cached_follower_topics(author_id)
```
Они позволяют эффективно кешировать и получать информацию о подписках, связях между авторами, темами и публикациями.
## Система инвалидации кеша
### 1. Прямая инвалидация
Система поддерживает два типа инвалидации кеша:
#### 1.1. Инвалидация по префиксу
```python
async def invalidate_cache_by_prefix(prefix)
```
Позволяет инвалидировать все ключи кеша, начинающиеся с указанного префикса. Используется в резолверах для инвалидации группы кешей при массовых изменениях.
#### 1.2. Точечная инвалидация
```python
async def invalidate_authors_cache(author_id=None)
async def invalidate_topics_cache(topic_id=None)
```
Эти функции позволяют инвалидировать кеш только для конкретной сущности, что снижает нагрузку на Redis и предотвращает ненужную потерю кешированных данных. Если ID сущности не указан, используется инвалидация по префиксу.
Примеры использования точечной инвалидации:
```python
# Инвалидация кеша только для автора с ID 123
await invalidate_authors_cache(123)
# Инвалидация кеша только для темы с ID 456
await invalidate_topics_cache(456)
```
### 2. Отложенная инвалидация
Модуль `revalidator.py` реализует систему отложенной инвалидации кеша через класс `CacheRevalidationManager`:
```python
class CacheRevalidationManager:
# ...
async def process_revalidation(self):
# ...
def mark_for_revalidation(self, entity_id, entity_type):
# ...
```
Менеджер ревалидации работает как асинхронный фоновый процесс, который периодически (по умолчанию каждые 5 минут) проверяет наличие сущностей для ревалидации.
Особенности реализации:
- Для авторов и тем используется поштучная ревалидация каждой записи
- Для шаутов и реакций используется батчевая обработка, с порогом в 10 элементов
- При достижении порога система переключается на инвалидацию коллекций вместо поштучной обработки
- Специальный флаг `all` позволяет запустить полную инвалидацию всех записей типа
### 3. Автоматическая инвалидация через триггеры
Модуль `triggers.py` регистрирует обработчики событий SQLAlchemy, которые автоматически отмечают сущности для ревалидации при изменении данных в базе:
```python
def events_register():
event.listen(Author, "after_update", mark_for_revalidation)
event.listen(Topic, "after_update", mark_for_revalidation)
# и другие...
```
Триггеры имеют следующие особенности:
- Реагируют на события вставки, обновления и удаления
- Отмечают затронутые сущности для отложенной ревалидации
- Учитывают связи между сущностями (например, при изменении темы обновляются связанные шауты)
## Предварительное кеширование
Модуль `precache.py` реализует предварительное кеширование часто используемых данных при старте приложения:
```python
async def precache_data():
# ...
```
Эта функция выполняется при запуске приложения и заполняет кеш данными, которые будут часто запрашиваться пользователями.
## Примеры использования
### Простое кеширование результата запроса
```python
async def get_topics_with_stats(limit=10, offset=0, by="title"):
# Формирование ключа кеша по конвенции
cache_key = f"topics:stats:limit={limit}:offset={offset}:sort={by}"
cached_data = await get_cached_data(cache_key)
if cached_data:
return cached_data
# Выполнение запроса к базе данных
result = ... # логика получения данных
await cache_data(cache_key, result, ttl=300)
return result
```
### Использование обобщенной функции cached_query
```python
async def get_topics_with_stats(limit=10, offset=0, by="title"):
async def fetch_data(limit, offset, by):
# Логика получения данных
return result
# Формирование ключа кеша по конвенции
cache_key = f"topics:stats:limit={limit}:offset={offset}:sort={by}"
return await cached_query(
cache_key,
fetch_data,
ttl=300,
limit=limit,
offset=offset,
by=by
)
```
### Точечная инвалидация кеша при изменении данных
```python
async def update_topic(topic_id, new_data):
# Обновление данных в базе
# ...
# Точечная инвалидация кеша только для измененной темы
await invalidate_topics_cache(topic_id)
return updated_topic
```
## Отладка и мониторинг
Система кеширования использует логгер для отслеживания операций:
```python
logger.debug(f"Данные получены из кеша по ключу {key}")
logger.debug(f"Удалено {len(keys)} ключей кеша с префиксом {prefix}")
logger.error(f"Ошибка при инвалидации кеша: {e}")
```
Это позволяет отслеживать работу кеша и выявлять возможные проблемы на ранних стадиях.
## Рекомендации по использованию
1. **Следуйте конвенциям формирования ключей** - это критически важно для консистентности и предсказуемости кеша.
2. **Не создавайте собственные форматы ключей** - используйте существующие шаблоны для обеспечения единообразия.
3. **Не забывайте об инвалидации** - всегда инвалидируйте кеш при изменении данных.
4. **Используйте точечную инвалидацию** - вместо инвалидации по префиксу для снижения нагрузки на Redis.
5. **Устанавливайте разумные TTL** - используйте разные значения TTL в зависимости от частоты изменения данных.
6. **Не кешируйте большие объемы данных** - кешируйте только то, что действительно необходимо для повышения производительности.
## Технические детали реализации
- **Сериализация данных**: используется `orjson` для эффективной сериализации и десериализации данных.
- **Форматирование даты и времени**: для корректной работы с датами используется `CustomJSONEncoder`.
- **Асинхронность**: все операции кеширования выполняются асинхронно для минимального влияния на производительность API.
- **Прямое взаимодействие с Redis**: все операции выполняются через прямые вызовы `redis.execute()` с обработкой ошибок.
- **Батчевая обработка**: для массовых операций используется пороговое значение, после которого применяются оптимизированные стратегии.
## Известные ограничения
1. **Согласованность данных** - система не гарантирует абсолютную согласованность данных в кеше и базе данных.
2. **Память** - необходимо следить за объемом данных в кеше, чтобы избежать проблем с памятью Redis.
3. **Производительность Redis** - при большом количестве операций с кешем может стать узким местом.

165
docs/comments-pagination.md Normal file
View File

@@ -0,0 +1,165 @@
# Пагинация комментариев
## Обзор
Реализована система пагинации комментариев по веткам, которая позволяет эффективно загружать и отображать вложенные ветки обсуждений. Основные преимущества:
1. Загрузка только необходимых комментариев, а не всего дерева
2. Снижение нагрузки на сервер и клиент
3. Возможность эффективной навигации по большим обсуждениям
4. Предзагрузка первых N ответов для улучшения UX
## API для иерархической загрузки комментариев
### GraphQL запрос `load_comments_branch`
```graphql
query LoadCommentsBranch(
$shout: Int!,
$parentId: Int,
$limit: Int,
$offset: Int,
$sort: ReactionSort,
$childrenLimit: Int,
$childrenOffset: Int
) {
load_comments_branch(
shout: $shout,
parent_id: $parentId,
limit: $limit,
offset: $offset,
sort: $sort,
children_limit: $childrenLimit,
children_offset: $childrenOffset
) {
id
body
created_at
created_by {
id
name
slug
pic
}
kind
reply_to
stat {
rating
commented
}
first_replies {
id
body
created_at
created_by {
id
name
slug
pic
}
kind
reply_to
stat {
rating
commented
}
}
}
}
```
### Параметры запроса
| Параметр | Тип | По умолчанию | Описание |
|----------|-----|--------------|----------|
| shout | Int! | - | ID статьи, к которой относятся комментарии |
| parent_id | Int | null | ID родительского комментария. Если null, загружаются корневые комментарии |
| limit | Int | 10 | Максимальное количество комментариев для загрузки |
| offset | Int | 0 | Смещение для пагинации |
| sort | ReactionSort | newest | Порядок сортировки: newest, oldest, like |
| children_limit | Int | 3 | Максимальное количество дочерних комментариев для каждого родительского |
| children_offset | Int | 0 | Смещение для пагинации дочерних комментариев |
### Поля в ответе
Каждый комментарий содержит следующие основные поля:
- `id`: ID комментария
- `body`: Текст комментария
- `created_at`: Время создания
- `created_by`: Информация об авторе
- `kind`: Тип реакции (COMMENT)
- `reply_to`: ID родительского комментария (null для корневых)
- `first_replies`: Первые N дочерних комментариев
- `stat`: Статистика комментария, включающая:
- `commented`: Количество ответов на комментарий
- `rating`: Рейтинг комментария
## Примеры использования
### Загрузка корневых комментариев с первыми ответами
```javascript
const { data } = await client.query({
query: LOAD_COMMENTS_BRANCH,
variables: {
shout: 222,
limit: 10,
offset: 0,
sort: "newest",
childrenLimit: 3
}
});
```
### Загрузка ответов на конкретный комментарий
```javascript
const { data } = await client.query({
query: LOAD_COMMENTS_BRANCH,
variables: {
shout: 222,
parentId: 123, // ID комментария, для которого загружаем ответы
limit: 10,
offset: 0,
sort: "oldest" // Сортируем ответы от старых к новым
}
});
```
### Пагинация дочерних комментариев
Для загрузки дополнительных ответов на комментарий:
```javascript
const { data } = await client.query({
query: LOAD_COMMENTS_BRANCH,
variables: {
shout: 222,
parentId: 123,
limit: 10,
offset: 0,
childrenLimit: 5,
childrenOffset: 3 // Пропускаем первые 3 комментария (уже загруженные)
}
});
```
## Рекомендации по клиентской реализации
1. Для эффективной работы со сложными ветками обсуждений рекомендуется:
- Сначала загружать только корневые комментарии с первыми N ответами
- При наличии дополнительных ответов (когда `stat.commented > first_replies.length`)
добавить кнопку "Показать все ответы"
- При нажатии на кнопку загружать дополнительные ответы с помощью запроса с указанным `parentId`
2. Для сортировки:
- По умолчанию использовать `newest` для отображения свежих обсуждений
- Предусмотреть переключатель сортировки для всего дерева комментариев
- При изменении сортировки перезагружать данные с новым параметром `sort`
3. Для улучшения производительности:
- Кешировать результаты запросов на клиенте
- Использовать оптимистичные обновления при добавлении/редактировании комментариев
- При необходимости загружать комментарии порциями (ленивая загрузка)

View File

@@ -6,14 +6,20 @@
## Мультидоменная авторизация
- Поддержка авторизации для разных доменов:
- *.dscrs.site (включая testing.dscrs.site)
- localhost[:port]
- testingdiscoursio-git-*-discoursio.vercel.app
- *.discours.io
- Поддержка авторизации для разных доменов
- Автоматическое определение сервера авторизации
- Корректная обработка CORS для всех поддерживаемых доменов
## Система кеширования
- Redis используется в качестве основного механизма кеширования
- Поддержка как синхронных, так и асинхронных функций в декораторе cache_on_arguments
- Автоматическая сериализация/десериализация данных в JSON с использованием CustomJSONEncoder
- Резервная сериализация через pickle для сложных объектов
- Генерация уникальных ключей кеша на основе сигнатуры функции и переданных аргументов
- Настраиваемое время жизни кеша (TTL)
- Возможность ручной инвалидации кеша для конкретных функций и аргументов
## Webhooks
- Автоматическая регистрация вебхука для события user.login
@@ -25,11 +31,18 @@
## CORS Configuration
- Поддерживаются домены:
- *.dscrs.site (включая testing.dscrs.site, core.dscrs.site)
- *.discours.io (включая testing.discours.io)
- localhost (включая порты)
- Поддерживаемые методы: GET, POST, OPTIONS
- Настроена поддержка credentials
- Разрешенные заголовки: Authorization, Content-Type, X-Requested-With, DNT, Cache-Control
- Настроено кэширование preflight-ответов на 20 дней (1728000 секунд)
- Настроено кэширование preflight-ответов на 20 дней (1728000 секунд)
## Пагинация комментариев по веткам
- Эффективная загрузка комментариев с учетом их иерархической структуры
- Отдельный запрос `load_comments_branch` для оптимизированной загрузки ветки комментариев
- Возможность загрузки корневых комментариев статьи с первыми ответами на них
- Гибкая пагинация как для корневых, так и для дочерних комментариев
- Использование поля `stat.commented` для отображения количества ответов на комментарий
- Добавление специального поля `first_replies` для хранения первых ответов на комментарий
- Поддержка различных методов сортировки (новые, старые, популярные)
- Оптимизированные SQL запросы для минимизации нагрузки на базу данных

48
main.py
View File

@@ -17,8 +17,7 @@ from cache.revalidator import revalidation_manager
from services.exception import ExceptionHandlerMiddleware
from services.redis import redis
from services.schema import create_all_tables, resolvers
#from services.search import search_service
from services.search import search_service, initialize_search_index
from services.search import search_service
from services.viewed import ViewedStorage
from services.webhook import WebhookEndpoint, create_webhook_endpoint
from settings import DEV_SERVER_PID_FILE_NAME, MODE
@@ -35,67 +34,24 @@ async def start():
f.write(str(os.getpid()))
print(f"[main] process started in {MODE} mode")
async def check_search_service():
"""Check if search service is available and log result"""
info = await search_service.info()
if info.get("status") in ["error", "unavailable"]:
print(f"[WARNING] Search service unavailable: {info.get('message', 'unknown reason')}")
else:
print(f"[INFO] Search service is available: {info}")
# indexing DB data
# async def indexing():
# from services.db import fetch_all_shouts
# all_shouts = await fetch_all_shouts()
# await initialize_search_index(all_shouts)
async def lifespan(_app):
try:
print("[lifespan] Starting application initialization")
create_all_tables()
await asyncio.gather(
redis.connect(),
precache_data(),
ViewedStorage.init(),
create_webhook_endpoint(),
check_search_service(),
search_service.info(),
start(),
revalidation_manager.start(),
)
print("[lifespan] Basic initialization complete")
# Add a delay before starting the intensive search indexing
print("[lifespan] Waiting for system stabilization before search indexing...")
await asyncio.sleep(10) # 10-second delay to let the system stabilize
# Start search indexing as a background task with lower priority
asyncio.create_task(initialize_search_index_background())
yield
finally:
print("[lifespan] Shutting down application services")
tasks = [redis.disconnect(), ViewedStorage.stop(), revalidation_manager.stop()]
await asyncio.gather(*tasks, return_exceptions=True)
print("[lifespan] Shutdown complete")
# Initialize search index in the background
async def initialize_search_index_background():
"""Run search indexing as a background task with low priority"""
try:
print("[search] Starting background search indexing process")
from services.db import fetch_all_shouts
# Get total count first (optional)
all_shouts = await fetch_all_shouts()
total_count = len(all_shouts) if all_shouts else 0
print(f"[search] Fetched {total_count} shouts for background indexing")
# Start the indexing process with the fetched shouts
print("[search] Beginning background search index initialization...")
await initialize_search_index(all_shouts)
print("[search] Background search index initialization complete")
except Exception as e:
print(f"[search] Error in background search indexing: {str(e)}")
# Создаем экземпляр GraphQL
graphql_app = GraphQL(schema, debug=True)

View File

@@ -1,6 +1,6 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
from services.db import Base
@@ -8,6 +8,15 @@ from services.db import Base
class AuthorRating(Base):
"""
Рейтинг автора от другого автора.
Attributes:
rater (int): ID оценивающего автора
author (int): ID оцениваемого автора
plus (bool): Положительная/отрицательная оценка
"""
__tablename__ = "author_rating"
id = None # type: ignore
@@ -15,8 +24,26 @@ class AuthorRating(Base):
author = Column(ForeignKey("author.id"), primary_key=True)
plus = Column(Boolean)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех оценок конкретного автора
Index("idx_author_rating_author", "author"),
# Индекс для быстрого поиска всех оценок, оставленных конкретным автором
Index("idx_author_rating_rater", "rater"),
)
class AuthorFollower(Base):
"""
Подписка одного автора на другого.
Attributes:
follower (int): ID подписчика
author (int): ID автора, на которого подписываются
created_at (int): Время создания подписки
auto (bool): Признак автоматической подписки
"""
__tablename__ = "author_follower"
id = None # type: ignore
@@ -25,16 +52,57 @@ class AuthorFollower(Base):
created_at = Column(Integer, nullable=False, default=lambda: int(time.time()))
auto = Column(Boolean, nullable=False, default=False)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех подписчиков автора
Index("idx_author_follower_author", "author"),
# Индекс для быстрого поиска всех авторов, на которых подписан конкретный автор
Index("idx_author_follower_follower", "follower"),
)
class AuthorBookmark(Base):
"""
Закладка автора на публикацию.
Attributes:
author (int): ID автора
shout (int): ID публикации
"""
__tablename__ = "author_bookmark"
id = None # type: ignore
author = Column(ForeignKey("author.id"), primary_key=True)
shout = Column(ForeignKey("shout.id"), primary_key=True)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех закладок автора
Index("idx_author_bookmark_author", "author"),
# Индекс для быстрого поиска всех авторов, добавивших публикацию в закладки
Index("idx_author_bookmark_shout", "shout"),
)
class Author(Base):
"""
Модель автора в системе.
Attributes:
user (str): Идентификатор пользователя в системе авторизации
name (str): Отображаемое имя
slug (str): Уникальный строковый идентификатор
bio (str): Краткая биография/статус
about (str): Полное описание
pic (str): URL изображения профиля
links (dict): Ссылки на социальные сети и сайты
created_at (int): Время создания профиля
last_seen (int): Время последнего посещения
updated_at (int): Время последнего обновления
deleted_at (int): Время удаления (если профиль удален)
"""
__tablename__ = "author"
user = Column(String) # unbounded link with authorizer's User type
@@ -53,3 +121,17 @@ class Author(Base):
# search_vector = Column(
# TSVectorType("name", "slug", "bio", "about", regconfig="pg_catalog.russian")
# )
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска по slug
Index("idx_author_slug", "slug"),
# Индекс для быстрого поиска по идентификатору пользователя
Index("idx_author_user", "user"),
# Индекс для фильтрации неудаленных авторов
Index("idx_author_deleted_at", "deleted_at", postgresql_where=deleted_at.is_(None)),
# Индекс для сортировки по времени создания (для новых авторов)
Index("idx_author_created_at", "created_at"),
# Индекс для сортировки по времени последнего посещения
Index("idx_author_last_seen", "last_seen"),
)

View File

@@ -1,6 +1,6 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
from sqlalchemy.orm import relationship
from orm.author import Author
@@ -10,6 +10,15 @@ from services.db import Base
class ShoutTopic(Base):
"""
Связь между публикацией и темой.
Attributes:
shout (int): ID публикации
topic (int): ID темы
main (bool): Признак основной темы
"""
__tablename__ = "shout_topic"
id = None # type: ignore
@@ -17,6 +26,12 @@ class ShoutTopic(Base):
topic = Column(ForeignKey("topic.id"), primary_key=True, index=True)
main = Column(Boolean, nullable=True)
# Определяем дополнительные индексы
__table_args__ = (
# Оптимизированный составной индекс для запросов, которые ищут публикации по теме
Index("idx_shout_topic_topic_shout", "topic", "shout"),
)
class ShoutReactionsFollower(Base):
__tablename__ = "shout_reactions_followers"
@@ -30,6 +45,15 @@ class ShoutReactionsFollower(Base):
class ShoutAuthor(Base):
"""
Связь между публикацией и автором.
Attributes:
shout (int): ID публикации
author (int): ID автора
caption (str): Подпись автора
"""
__tablename__ = "shout_author"
id = None # type: ignore
@@ -37,8 +61,18 @@ class ShoutAuthor(Base):
author = Column(ForeignKey("author.id"), primary_key=True, index=True)
caption = Column(String, nullable=True, default="")
# Определяем дополнительные индексы
__table_args__ = (
# Оптимизированный индекс для запросов, которые ищут публикации по автору
Index("idx_shout_author_author_shout", "author", "shout"),
)
class Shout(Base):
"""
Публикация в системе.
"""
__tablename__ = "shout"
created_at: int = Column(Integer, nullable=False, default=lambda: int(time.time()))
@@ -74,3 +108,20 @@ class Shout(Base):
seo: str | None = Column(String, nullable=True) # JSON
draft: int | None = Column(ForeignKey("draft.id"), nullable=True)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска неудаленных публикаций
Index("idx_shout_deleted_at", "deleted_at", postgresql_where=deleted_at.is_(None)),
# Индекс для быстрой фильтрации по community
Index("idx_shout_community", "community"),
# Индекс для быстрого поиска по slug
Index("idx_shout_slug", "slug"),
# Составной индекс для фильтрации опубликованных неудаленных публикаций
Index(
"idx_shout_published_deleted",
"published_at",
"deleted_at",
postgresql_where=published_at.is_not(None) & deleted_at.is_(None),
),
)

View File

@@ -1,11 +1,21 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
from services.db import Base
class TopicFollower(Base):
"""
Связь между топиком и его подписчиком.
Attributes:
follower (int): ID подписчика
topic (int): ID топика
created_at (int): Время создания связи
auto (bool): Автоматическая подписка
"""
__tablename__ = "topic_followers"
id = None # type: ignore
@@ -14,8 +24,29 @@ class TopicFollower(Base):
created_at = Column(Integer, nullable=False, default=int(time.time()))
auto = Column(Boolean, nullable=False, default=False)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех подписчиков топика
Index("idx_topic_followers_topic", "topic"),
# Индекс для быстрого поиска всех топиков, на которые подписан автор
Index("idx_topic_followers_follower", "follower"),
)
class Topic(Base):
"""
Модель топика (темы) публикаций.
Attributes:
slug (str): Уникальный строковый идентификатор темы
title (str): Название темы
body (str): Описание темы
pic (str): URL изображения темы
community (int): ID сообщества
oid (str): Старый ID
parent_ids (list): IDs родительских тем
"""
__tablename__ = "topic"
slug = Column(String, unique=True)
@@ -24,5 +55,12 @@ class Topic(Base):
pic = Column(String, nullable=True, comment="Picture")
community = Column(ForeignKey("community.id"), default=1)
oid = Column(String, nullable=True, comment="Old ID")
parent_ids = Column(JSON, nullable=True, comment="Parent Topic IDs")
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска по slug
Index("idx_topic_slug", "slug"),
# Индекс для быстрого поиска по сообществу
Index("idx_topic_community", "community"),
)

6
requirements.dev.txt Normal file
View File

@@ -0,0 +1,6 @@
fakeredis
pytest
pytest-asyncio
pytest-cov
mypy
ruff

View File

@@ -2,13 +2,10 @@
bcrypt
authlib
passlib
google-analytics-data
dogpile-cache
opensearch-py
google-analytics-data
colorlog
psycopg2-binary
dogpile-cache
httpx
redis[hiredis]
sentry-sdk[starlette,sqlalchemy]
@@ -16,14 +13,5 @@ starlette
gql
ariadne
granian
# NLP and search
httpx
pydantic
fakeredis
pytest
pytest-asyncio
pytest-cov
mypy
ruff
orjson
pydantic

View File

@@ -37,6 +37,7 @@ from resolvers.reaction import (
create_reaction,
delete_reaction,
load_comment_ratings,
load_comments_branch,
load_reactions_by,
load_shout_comments,
load_shout_ratings,
@@ -107,6 +108,7 @@ __all__ = [
"load_shout_comments",
"load_shout_ratings",
"load_comment_ratings",
"load_comments_branch",
# notifier
"load_notifications",
"notifications_seen_thread",

View File

@@ -1,25 +1,196 @@
import asyncio
import time
from typing import Optional
from sqlalchemy import desc, select, text
from sqlalchemy import select, text
from cache.cache import (
cache_author,
cached_query,
get_cached_author,
get_cached_author_by_user_id,
get_cached_author_followers,
get_cached_follower_authors,
get_cached_follower_topics,
invalidate_cache_by_prefix,
)
from orm.author import Author
from orm.shout import ShoutAuthor, ShoutTopic
from orm.topic import Topic
from resolvers.stat import get_with_stat
from services.auth import login_required
from services.db import local_session
from services.redis import redis
from services.schema import mutation, query
from utils.logger import root_logger as logger
DEFAULT_COMMUNITIES = [1]
# Вспомогательная функция для получения всех авторов без статистики
async def get_all_authors():
"""
Получает всех авторов без статистики.
Используется для случаев, когда нужен полный список авторов без дополнительной информации.
Returns:
list: Список всех авторов без статистики
"""
cache_key = "authors:all:basic"
# Функция для получения всех авторов из БД
async def fetch_all_authors():
logger.debug("Получаем список всех авторов из БД и кешируем результат")
with local_session() as session:
# Запрос на получение базовой информации об авторах
authors_query = select(Author).where(Author.deleted_at.is_(None))
authors = session.execute(authors_query).scalars().all()
# Преобразуем авторов в словари
return [author.dict() for author in authors]
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_all_authors)
# Вспомогательная функция для получения авторов со статистикой с пагинацией
async def get_authors_with_stats(limit=50, offset=0, by: Optional[str] = None):
"""
Получает авторов со статистикой с пагинацией.
Args:
limit: Максимальное количество возвращаемых авторов
offset: Смещение для пагинации
by: Опциональный параметр сортировки (new/active)
Returns:
list: Список авторов с их статистикой
"""
# Формируем ключ кеша с помощью универсальной функции
cache_key = f"authors:stats:limit={limit}:offset={offset}"
# Функция для получения авторов из БД
async def fetch_authors_with_stats():
logger.debug(f"Выполняем запрос на получение авторов со статистикой: limit={limit}, offset={offset}, by={by}")
with local_session() as session:
# Базовый запрос для получения авторов
base_query = select(Author).where(Author.deleted_at.is_(None))
# Применяем сортировку
if by:
if isinstance(by, dict):
# Обработка словаря параметров сортировки
from sqlalchemy import asc, desc
for field, direction in by.items():
column = getattr(Author, field, None)
if column:
if direction.lower() == "desc":
base_query = base_query.order_by(desc(column))
else:
base_query = base_query.order_by(column)
elif by == "new":
base_query = base_query.order_by(desc(Author.created_at))
elif by == "active":
base_query = base_query.order_by(desc(Author.last_seen))
else:
# По умолчанию сортируем по времени создания
base_query = base_query.order_by(desc(Author.created_at))
else:
base_query = base_query.order_by(desc(Author.created_at))
# Применяем лимит и смещение
base_query = base_query.limit(limit).offset(offset)
# Получаем авторов
authors = session.execute(base_query).scalars().all()
author_ids = [author.id for author in authors]
if not author_ids:
return []
# Оптимизированный запрос для получения статистики по публикациям для авторов
shouts_stats_query = f"""
SELECT sa.author, COUNT(DISTINCT s.id) as shouts_count
FROM shout_author sa
JOIN shout s ON sa.shout = s.id AND s.deleted_at IS NULL AND s.published_at IS NOT NULL
WHERE sa.author IN ({",".join(map(str, author_ids))})
GROUP BY sa.author
"""
shouts_stats = {row[0]: row[1] for row in session.execute(text(shouts_stats_query))}
# Запрос на получение статистики по подписчикам для авторов
followers_stats_query = f"""
SELECT author, COUNT(DISTINCT follower) as followers_count
FROM author_follower
WHERE author IN ({",".join(map(str, author_ids))})
GROUP BY author
"""
followers_stats = {row[0]: row[1] for row in session.execute(text(followers_stats_query))}
# Формируем результат с добавлением статистики
result = []
for author in authors:
author_dict = author.dict()
author_dict["stat"] = {
"shouts": shouts_stats.get(author.id, 0),
"followers": followers_stats.get(author.id, 0),
}
result.append(author_dict)
# Кешируем каждого автора отдельно для использования в других функциях
await cache_author(author_dict)
return result
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_authors_with_stats)
# Функция для инвалидации кеша авторов
async def invalidate_authors_cache(author_id=None):
"""
Инвалидирует кеши авторов при изменении данных.
Args:
author_id: Опциональный ID автора для точечной инвалидации.
Если не указан, инвалидируются все кеши авторов.
"""
if author_id:
# Точечная инвалидация конкретного автора
logger.debug(f"Инвалидация кеша для автора #{author_id}")
specific_keys = [
f"author:id:{author_id}",
f"author:followers:{author_id}",
f"author:follows-authors:{author_id}",
f"author:follows-topics:{author_id}",
f"author:follows-shouts:{author_id}",
]
# Получаем user_id автора, если есть
with local_session() as session:
author = session.query(Author).filter(Author.id == author_id).first()
if author and author.user:
specific_keys.append(f"author:user:{author.user.strip()}")
# Удаляем конкретные ключи
for key in specific_keys:
try:
await redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
except Exception as e:
logger.error(f"Ошибка при удалении ключа {key}: {e}")
# Также ищем и удаляем ключи коллекций, содержащих данные об этом авторе
collection_keys = await redis.execute("KEYS", "authors:stats:*")
if collection_keys:
await redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей авторов")
else:
# Общая инвалидация всех кешей авторов
logger.debug("Полная инвалидация кеша авторов")
await invalidate_cache_by_prefix("authors")
@mutation.field("update_author")
@login_required
@@ -51,10 +222,30 @@ async def update_author(_, info, profile):
@query.field("get_authors_all")
def get_authors_all(_, _info):
with local_session() as session:
authors = session.query(Author).all()
return authors
async def get_authors_all(_, _info):
"""
Получает список всех авторов без статистики.
Returns:
list: Список всех авторов
"""
return await get_all_authors()
@query.field("get_authors_paginated")
async def get_authors_paginated(_, _info, limit=50, offset=0, by=None):
"""
Получает список авторов с пагинацией и статистикой.
Args:
limit: Максимальное количество возвращаемых авторов
offset: Смещение для пагинации
by: Параметр сортировки (new/active)
Returns:
list: Список авторов с их статистикой
"""
return await get_authors_with_stats(limit, offset, by)
@query.field("get_author")
@@ -105,145 +296,105 @@ async def get_author_id(_, _info, user: str):
asyncio.create_task(cache_author(author_dict))
return author_with_stat
except Exception as exc:
import traceback
traceback.print_exc()
logger.error(exc)
logger.error(f"Error getting author: {exc}")
return None
@query.field("load_authors_by")
async def load_authors_by(_, _info, by, limit, offset):
logger.debug(f"loading authors by {by}")
authors_query = select(Author)
"""
Загружает авторов по заданному критерию с пагинацией.
if by.get("slug"):
authors_query = authors_query.filter(Author.slug.ilike(f"%{by['slug']}%"))
elif by.get("name"):
authors_query = authors_query.filter(Author.name.ilike(f"%{by['name']}%"))
elif by.get("topic"):
authors_query = (
authors_query.join(ShoutAuthor) # Первое соединение ShoutAuthor
.join(ShoutTopic, ShoutAuthor.shout == ShoutTopic.shout)
.join(Topic, ShoutTopic.topic == Topic.id)
.filter(Topic.slug == str(by["topic"]))
)
Args:
by: Критерий сортировки авторов (new/active)
limit: Максимальное количество возвращаемых авторов
offset: Смещение для пагинации
if by.get("last_seen"): # в unix time
before = int(time.time()) - by["last_seen"]
authors_query = authors_query.filter(Author.last_seen > before)
elif by.get("created_at"): # в unix time
before = int(time.time()) - by["created_at"]
authors_query = authors_query.filter(Author.created_at > before)
authors_query = authors_query.limit(limit).offset(offset)
with local_session() as session:
authors_nostat = session.execute(authors_query).all()
authors = []
for a in authors_nostat:
if isinstance(a, Author):
author_dict = await get_cached_author(a.id, get_with_stat)
if author_dict and isinstance(author_dict.get("shouts"), int):
authors.append(author_dict)
# order
order = by.get("order")
if order in ["shouts", "followers"]:
authors_query = authors_query.order_by(desc(text(f"{order}_stat")))
# group by
authors = get_with_stat(authors_query)
return authors or []
Returns:
list: Список авторов с учетом критерия
"""
# Используем оптимизированную функцию для получения авторов
return await get_authors_with_stats(limit, offset, by)
def get_author_id_from(slug="", user=None, author_id=None):
if not slug and not user and not author_id:
raise ValueError("One of slug, user, or author_id must be provided")
author_query = select(Author.id)
if user:
author_query = author_query.filter(Author.user == user)
elif slug:
author_query = author_query.filter(Author.slug == slug)
elif author_id:
author_query = author_query.filter(Author.id == author_id)
with local_session() as session:
author_id_result = session.execute(author_query).first()
author_id = author_id_result[0] if author_id_result else None
if not author_id:
raise ValueError("Author not found")
try:
author_id = None
if author_id:
return author_id
with local_session() as session:
author = None
if slug:
author = session.query(Author).filter(Author.slug == slug).first()
if author:
author_id = author.id
return author_id
if user:
author = session.query(Author).filter(Author.user == user).first()
if author:
author_id = author.id
except Exception as exc:
logger.error(exc)
return author_id
@query.field("get_author_follows")
async def get_author_follows(_, _info, slug="", user=None, author_id=0):
try:
author_id = get_author_id_from(slug, user, author_id)
logger.debug(f"getting follows for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return {}
if bool(author_id):
logger.debug(f"getting {author_id} follows authors")
authors = await get_cached_follower_authors(author_id)
topics = await get_cached_follower_topics(author_id)
return {
"topics": topics,
"authors": authors,
"communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
}
except Exception:
import traceback
followed_authors = await get_cached_follower_authors(author_id)
followed_topics = await get_cached_follower_topics(author_id)
traceback.print_exc()
return {"error": "Author not found"}
# TODO: Get followed communities too
return {
"authors": followed_authors,
"topics": followed_topics,
"communities": DEFAULT_COMMUNITIES,
"shouts": [],
}
@query.field("get_author_follows_topics")
async def get_author_follows_topics(_, _info, slug="", user=None, author_id=None):
try:
follower_id = get_author_id_from(slug, user, author_id)
topics = await get_cached_follower_topics(follower_id)
return topics
except Exception:
import traceback
traceback.print_exc()
logger.debug(f"getting followed topics for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return []
followed_topics = await get_cached_follower_topics(author_id)
return followed_topics
@query.field("get_author_follows_authors")
async def get_author_follows_authors(_, _info, slug="", user=None, author_id=None):
try:
follower_id = get_author_id_from(slug, user, author_id)
return await get_cached_follower_authors(follower_id)
except Exception:
import traceback
traceback.print_exc()
logger.debug(f"getting followed authors for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return []
followed_authors = await get_cached_follower_authors(author_id)
return followed_authors
def create_author(user_id: str, slug: str, name: str = ""):
author = Author()
author.user = user_id # Связь с user_id из системы авторизации
author.slug = slug # Идентификатор из системы авторизации
author.created_at = author.updated_at = int(time.time())
author.name = name or slug # если не указано
with local_session() as session:
try:
author = None
if user_id:
author = session.query(Author).filter(Author.user == user_id).first()
elif slug:
author = session.query(Author).filter(Author.slug == slug).first()
if not author:
new_author = Author(user=user_id, slug=slug, name=name)
session.add(new_author)
session.commit()
logger.info(f"author created by webhook {new_author.dict()}")
except Exception as exc:
logger.debug(exc)
session.add(author)
session.commit()
return author
@query.field("get_author_followers")
async def get_author_followers(_, _info, slug: str = "", user: str = "", author_id: int = 0):
logger.debug(f"getting followers for @{slug}")
logger.debug(f"getting followers for author @{slug} or ID:{author_id}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
followers = []
if author_id:
followers = await get_cached_author_followers(author_id)
if not author_id:
return []
followers = await get_cached_author_followers(author_id)
return followers

View File

@@ -1,4 +1,5 @@
import time
from operator import or_
from sqlalchemy.sql import and_
@@ -55,7 +56,11 @@ async def load_drafts(_, info):
return {"error": "User ID and author ID are required"}
with local_session() as session:
drafts = session.query(Draft).filter(Draft.authors.any(Author.id == author_id)).all()
drafts = (
session.query(Draft)
.filter(or_(Draft.authors.any(Author.id == author_id), Draft.created_by == author_id))
.all()
)
return {"drafts": drafts}
@@ -96,7 +101,7 @@ async def create_draft(_, info, draft_input):
# Проверяем обязательные поля
if "body" not in draft_input or not draft_input["body"]:
draft_input["body"] = "" # Пустая строка вместо NULL
if "title" not in draft_input or not draft_input["title"]:
draft_input["title"] = "" # Пустая строка вместо NULL
@@ -120,23 +125,29 @@ async def create_draft(_, info, draft_input):
@mutation.field("update_draft")
@login_required
async def update_draft(_, info, draft_input):
async def update_draft(_, info, draft_id: int, draft_input):
"""Обновляет черновик публикации.
Args:
draft_id: ID черновика для обновления
draft_input: Данные для обновления черновика
Returns:
dict: Обновленный черновик или сообщение об ошибке
"""
user_id = info.context.get("user_id")
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
draft_id = draft_input.get("id")
if not draft_id:
return {"error": "Draft ID is required"}
if not user_id or not author_id:
return {"error": "Author ID are required"}
with local_session() as session:
draft = session.query(Draft).filter(Draft.id == draft_id).first()
del draft_input["id"]
Draft.update(draft, {**draft_input})
if not draft:
return {"error": "Draft not found"}
Draft.update(draft, draft_input)
draft.updated_at = int(time.time())
session.commit()
return {"draft": draft}

View File

@@ -1,6 +1,6 @@
import json
import time
import orjson
from sqlalchemy import and_, desc, select
from sqlalchemy.orm import joinedload
from sqlalchemy.sql.functions import coalesce
@@ -106,7 +106,7 @@ async def get_my_shout(_, info, shout_id: int):
if hasattr(shout, "media") and shout.media:
if isinstance(shout.media, str):
try:
shout.media = json.loads(shout.media)
shout.media = orjson.loads(shout.media)
except Exception as e:
logger.error(f"Error parsing shout media: {e}")
shout.media = []

View File

@@ -1,7 +1,7 @@
import json
import time
from typing import List, Tuple
import orjson
from sqlalchemy import and_, select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import aliased
@@ -115,7 +115,7 @@ def get_notifications_grouped(author_id: int, after: int = 0, limit: int = 10, o
if (groups_amount + offset) >= limit:
break
payload = json.loads(str(notification.payload))
payload = orjson.loads(str(notification.payload))
if str(notification.entity) == NotificationEntity.SHOUT.value:
shout = payload
@@ -177,7 +177,7 @@ def get_notifications_grouped(author_id: int, after: int = 0, limit: int = 10, o
elif str(notification.entity) == "follower":
thread_id = "followers"
follower = json.loads(payload)
follower = orjson.loads(payload)
group = groups_by_thread.get(thread_id)
if group:
if str(notification.action) == "follow":
@@ -293,11 +293,11 @@ async def notifications_seen_thread(_, info, thread: str, after: int):
)
exclude = set()
for nr in removed_reaction_notifications:
reaction = json.loads(str(nr.payload))
reaction = orjson.loads(str(nr.payload))
reaction_id = reaction.get("id")
exclude.add(reaction_id)
for n in new_reaction_notifications:
reaction = json.loads(str(n.payload))
reaction = orjson.loads(str(n.payload))
reaction_id = reaction.get("id")
if (
reaction_id not in exclude

View File

@@ -97,20 +97,23 @@ def get_reactions_with_stat(q, limit, offset):
def is_featured_author(session, author_id) -> bool:
"""
Check if an author has at least one featured article.
Check if an author has at least one non-deleted featured article.
:param session: Database session.
:param author_id: Author ID.
:return: True if the author has a featured article, else False.
"""
return session.query(
session.query(Shout).where(Shout.authors.any(id=author_id)).filter(Shout.featured_at.is_not(None)).exists()
session.query(Shout)
.where(Shout.authors.any(id=author_id))
.filter(Shout.featured_at.is_not(None), Shout.deleted_at.is_(None))
.exists()
).scalar()
def check_to_feature(session, approver_id, reaction) -> bool:
"""
Make a shout featured if it receives more than 4 votes.
Make a shout featured if it receives more than 4 votes from authors.
:param session: Database session.
:param approver_id: Approver author ID.
@@ -118,46 +121,78 @@ def check_to_feature(session, approver_id, reaction) -> bool:
:return: True if shout should be featured, else False.
"""
if not reaction.reply_to and is_positive(reaction.kind):
approvers = {approver_id}
# Count the number of approvers
# Проверяем, не содержит ли пост более 20% дизлайков
# Если да, то не должен быть featured независимо от количества лайков
if check_to_unfeature(session, reaction):
return False
# Собираем всех авторов, поставивших лайк
author_approvers = set()
reacted_readers = (
session.query(Reaction.created_by)
.filter(Reaction.shout == reaction.shout, is_positive(Reaction.kind), Reaction.deleted_at.is_(None))
.filter(
Reaction.shout == reaction.shout,
is_positive(Reaction.kind),
# Рейтинги (LIKE, DISLIKE) физически удаляются, поэтому фильтр deleted_at не нужен
)
.distinct()
.all()
)
for reader_id in reacted_readers:
# Добавляем текущего одобряющего
approver = session.query(Author).filter(Author.id == approver_id).first()
if approver and is_featured_author(session, approver_id):
author_approvers.add(approver_id)
# Проверяем, есть ли у реагировавших авторов featured публикации
for (reader_id,) in reacted_readers:
if is_featured_author(session, reader_id):
approvers.add(reader_id)
return len(approvers) > 4
author_approvers.add(reader_id)
# Публикация становится featured при наличии более 4 лайков от авторов
logger.debug(f"Публикация {reaction.shout} имеет {len(author_approvers)} лайков от авторов")
return len(author_approvers) > 4
return False
def check_to_unfeature(session, rejecter_id, reaction) -> bool:
def check_to_unfeature(session, reaction) -> bool:
"""
Unfeature a shout if 20% of reactions are negative.
:param session: Database session.
:param rejecter_id: Rejecter author ID.
:param reaction: Reaction object.
:return: True if shout should be unfeatured, else False.
"""
if not reaction.reply_to and is_negative(reaction.kind):
if not reaction.reply_to:
# Проверяем соотношение дизлайков, даже если текущая реакция не дизлайк
total_reactions = (
session.query(Reaction)
.filter(
Reaction.shout == reaction.shout, Reaction.kind.in_(RATING_REACTIONS), Reaction.deleted_at.is_(None)
Reaction.shout == reaction.shout,
Reaction.reply_to.is_(None),
Reaction.kind.in_(RATING_REACTIONS),
# Рейтинги физически удаляются при удалении, поэтому фильтр deleted_at не нужен
)
.count()
)
negative_reactions = (
session.query(Reaction)
.filter(Reaction.shout == reaction.shout, is_negative(Reaction.kind), Reaction.deleted_at.is_(None))
.filter(
Reaction.shout == reaction.shout,
is_negative(Reaction.kind),
Reaction.reply_to.is_(None),
# Рейтинги физически удаляются при удалении, поэтому фильтр deleted_at не нужен
)
.count()
)
return total_reactions > 0 and (negative_reactions / total_reactions) >= 0.2
# Проверяем, составляют ли отрицательные реакции 20% или более от всех реакций
negative_ratio = negative_reactions / total_reactions if total_reactions > 0 else 0
logger.debug(
f"Публикация {reaction.shout}: {negative_reactions}/{total_reactions} отрицательных реакций ({negative_ratio:.2%})"
)
return total_reactions > 0 and negative_ratio >= 0.2
return False
@@ -196,8 +231,8 @@ async def _create_reaction(session, shout_id: int, is_author: bool, author_id: i
Create a new reaction and perform related actions such as updating counters and notification.
:param session: Database session.
:param info: GraphQL context info.
:param shout: Shout object.
:param shout_id: Shout ID.
:param is_author: Flag indicating if the user is the author of the shout.
:param author_id: Author ID.
:param reaction: Dictionary with reaction data.
:return: Dictionary with created reaction data.
@@ -217,10 +252,14 @@ async def _create_reaction(session, shout_id: int, is_author: bool, author_id: i
# Handle rating
if r.kind in RATING_REACTIONS:
if check_to_unfeature(session, author_id, r):
# Проверяем сначала условие для unfeature (дизлайки имеют приоритет)
if check_to_unfeature(session, r):
set_unfeatured(session, shout_id)
logger.info(f"Публикация {shout_id} потеряла статус featured из-за высокого процента дизлайков")
# Только если не было unfeature, проверяем условие для feature
elif check_to_feature(session, author_id, r):
await set_featured(session, shout_id)
logger.info(f"Публикация {shout_id} получила статус featured благодаря лайкам от авторов")
# Notify creation
await notify_reaction(rdict, "create")
@@ -354,7 +393,7 @@ async def update_reaction(_, info, reaction):
result = session.execute(reaction_query).unique().first()
if result:
r, author, shout, commented_stat, rating_stat = result
r, author, _shout, commented_stat, rating_stat = result
if not r or not author:
return {"error": "Invalid reaction ID or unauthorized"}
@@ -406,15 +445,24 @@ async def delete_reaction(_, info, reaction_id: int):
if r.created_by != author_id and "editor" not in roles:
return {"error": "Access denied"}
logger.debug(f"{user_id} user removing his #{reaction_id} reaction")
reaction_dict = r.dict()
session.delete(r)
session.commit()
# Update author stat
if r.kind == ReactionKind.COMMENT.value:
r.deleted_at = int(time.time())
update_author_stat(author.id)
session.add(r)
session.commit()
elif r.kind == ReactionKind.PROPOSE.value:
r.deleted_at = int(time.time())
session.add(r)
session.commit()
# TODO: add more reaction types here
else:
logger.debug(f"{user_id} user removing his #{reaction_id} reaction")
session.delete(r)
session.commit()
if check_to_unfeature(session, r):
set_unfeatured(session, r.shout)
reaction_dict = r.dict()
await notify_reaction(reaction_dict, "delete")
return {"error": None, "reaction": reaction_dict}
@@ -485,7 +533,9 @@ async def load_reactions_by(_, _info, by, limit=50, offset=0):
# Add statistics and apply filters
q = add_reaction_stat_columns(q)
q = apply_reaction_filters(by, q)
q = q.where(Reaction.deleted_at.is_(None))
# Include reactions with deleted_at for building comment trees
# q = q.where(Reaction.deleted_at.is_(None))
# Group and sort
q = q.group_by(Reaction.id, Author.id, Shout.id)
@@ -562,24 +612,22 @@ async def load_shout_comments(_, info, shout: int, limit=50, offset=0):
@query.field("load_comment_ratings")
async def load_comment_ratings(_, info, comment: int, limit=50, offset=0):
"""
Load ratings for a specified comment with pagination and statistics.
Load ratings for a specified comment with pagination.
:param info: GraphQL context info.
:param comment: Comment ID.
:param limit: Number of ratings to load.
:param offset: Pagination offset.
:return: List of reactions.
:return: List of ratings.
"""
q = query_reactions()
q = add_reaction_stat_columns(q)
# Filter, group, sort, limit, offset
q = q.filter(
and_(
Reaction.deleted_at.is_(None),
Reaction.reply_to == comment,
Reaction.kind == ReactionKind.COMMENT.value,
Reaction.kind.in_(RATING_REACTIONS),
)
)
q = q.group_by(Reaction.id, Author.id, Shout.id)
@@ -587,3 +635,186 @@ async def load_comment_ratings(_, info, comment: int, limit=50, offset=0):
# Retrieve and return reactions
return get_reactions_with_stat(q, limit, offset)
@query.field("load_comments_branch")
async def load_comments_branch(
_,
_info,
shout: int,
parent_id: int | None = None,
limit=10,
offset=0,
sort="newest",
children_limit=3,
children_offset=0,
):
"""
Загружает иерархические комментарии с возможностью пагинации корневых и дочерних.
:param info: GraphQL context info.
:param shout: ID статьи.
:param parent_id: ID родительского комментария (None для корневых).
:param limit: Количество комментариев для загрузки.
:param offset: Смещение для пагинации.
:param sort: Порядок сортировки ('newest', 'oldest', 'like').
:param children_limit: Максимальное количество дочерних комментариев.
:param children_offset: Смещение для дочерних комментариев.
:return: Список комментариев с дочерними.
"""
# Создаем базовый запрос
q = query_reactions()
q = add_reaction_stat_columns(q)
# Фильтруем по статье и типу (комментарии)
q = q.filter(
and_(
Reaction.deleted_at.is_(None),
Reaction.shout == shout,
Reaction.kind == ReactionKind.COMMENT.value,
)
)
# Фильтруем по родительскому ID
if parent_id is None:
# Загружаем только корневые комментарии
q = q.filter(Reaction.reply_to.is_(None))
else:
# Загружаем только прямые ответы на указанный комментарий
q = q.filter(Reaction.reply_to == parent_id)
# Сортировка и группировка
q = q.group_by(Reaction.id, Author.id, Shout.id)
# Определяем сортировку
order_by_stmt = None
if sort.lower() == "oldest":
order_by_stmt = asc(Reaction.created_at)
elif sort.lower() == "like":
order_by_stmt = desc("rating_stat")
else: # "newest" по умолчанию
order_by_stmt = desc(Reaction.created_at)
q = q.order_by(order_by_stmt)
# Выполняем запрос для получения комментариев
comments = get_reactions_with_stat(q, limit, offset)
# Если комментарии найдены, загружаем дочерние и количество ответов
if comments:
# Загружаем количество ответов для каждого комментария
await load_replies_count(comments)
# Загружаем дочерние комментарии
await load_first_replies(comments, children_limit, children_offset, sort)
return comments
async def load_replies_count(comments):
"""
Загружает количество ответов для списка комментариев и обновляет поле stat.commented.
:param comments: Список комментариев, для которых нужно загрузить количество ответов.
"""
if not comments:
return
comment_ids = [comment["id"] for comment in comments]
# Запрос для подсчета количества ответов
q = (
select(Reaction.reply_to.label("parent_id"), func.count().label("count"))
.where(
and_(
Reaction.reply_to.in_(comment_ids),
Reaction.deleted_at.is_(None),
Reaction.kind == ReactionKind.COMMENT.value,
)
)
.group_by(Reaction.reply_to)
)
# Выполняем запрос
with local_session() as session:
result = session.execute(q).fetchall()
# Создаем словарь {parent_id: count}
replies_count = {row[0]: row[1] for row in result}
# Добавляем значения в комментарии
for comment in comments:
if "stat" not in comment:
comment["stat"] = {}
# Обновляем счетчик комментариев в stat
comment["stat"]["commented"] = replies_count.get(comment["id"], 0)
async def load_first_replies(comments, limit, offset, sort="newest"):
"""
Загружает первые N ответов для каждого комментария.
:param comments: Список комментариев, для которых нужно загрузить ответы.
:param limit: Максимальное количество ответов для каждого комментария.
:param offset: Смещение для пагинации дочерних комментариев.
:param sort: Порядок сортировки ответов.
"""
if not comments or limit <= 0:
return
# Собираем ID комментариев
comment_ids = [comment["id"] for comment in comments]
# Базовый запрос для загрузки ответов
q = query_reactions()
q = add_reaction_stat_columns(q)
# Фильтрация: только ответы на указанные комментарии
q = q.filter(
and_(
Reaction.reply_to.in_(comment_ids),
Reaction.deleted_at.is_(None),
Reaction.kind == ReactionKind.COMMENT.value,
)
)
# Группировка
q = q.group_by(Reaction.id, Author.id, Shout.id)
# Определяем сортировку
order_by_stmt = None
if sort.lower() == "oldest":
order_by_stmt = asc(Reaction.created_at)
elif sort.lower() == "like":
order_by_stmt = desc("rating_stat")
else: # "newest" по умолчанию
order_by_stmt = desc(Reaction.created_at)
q = q.order_by(order_by_stmt, Reaction.reply_to)
# Выполняем запрос
replies = get_reactions_with_stat(q)
# Группируем ответы по родительским ID
replies_by_parent = {}
for reply in replies:
parent_id = reply.get("reply_to")
if parent_id not in replies_by_parent:
replies_by_parent[parent_id] = []
replies_by_parent[parent_id].append(reply)
# Добавляем ответы к соответствующим комментариям с учетом смещения и лимита
for comment in comments:
comment_id = comment["id"]
if comment_id in replies_by_parent:
parent_replies = replies_by_parent[comment_id]
# Применяем смещение и лимит
comment["first_replies"] = parent_replies[offset : offset + limit]
else:
comment["first_replies"] = []
# Загружаем количество ответов для дочерних комментариев
all_replies = [reply for replies in replies_by_parent.values() for reply in replies]
if all_replies:
await load_replies_count(all_replies)

View File

@@ -1,5 +1,4 @@
import json
import orjson
from graphql import GraphQLResolveInfo
from sqlalchemy import and_, nulls_last, text
from sqlalchemy.orm import aliased
@@ -222,7 +221,7 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
if has_field(info, "stat"):
stat = {}
if isinstance(row.stat, str):
stat = json.loads(row.stat)
stat = orjson.loads(row.stat)
elif isinstance(row.stat, dict):
stat = row.stat
viewed = ViewedStorage.get_shout(shout_id=shout_id) or 0
@@ -231,7 +230,7 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
# Обработка main_topic и topics
topics = None
if has_field(info, "topics") and hasattr(row, "topics"):
topics = json.loads(row.topics) if isinstance(row.topics, str) else row.topics
topics = orjson.loads(row.topics) if isinstance(row.topics, str) else row.topics
# logger.debug(f"Shout#{shout_id} topics: {topics}")
shout_dict["topics"] = topics
@@ -240,7 +239,7 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
if hasattr(row, "main_topic"):
# logger.debug(f"Raw main_topic for shout#{shout_id}: {row.main_topic}")
main_topic = (
json.loads(row.main_topic) if isinstance(row.main_topic, str) else row.main_topic
orjson.loads(row.main_topic) if isinstance(row.main_topic, str) else row.main_topic
)
# logger.debug(f"Parsed main_topic for shout#{shout_id}: {main_topic}")
@@ -253,14 +252,14 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
"is_main": True,
}
elif not main_topic:
logger.debug(f"No main_topic and no topics found for shout#{shout_id}")
logger.warning(f"No main_topic and no topics found for shout#{shout_id}")
main_topic = {"id": 0, "title": "no topic", "slug": "notopic", "is_main": True}
shout_dict["main_topic"] = main_topic
logger.debug(f"Final main_topic for shout#{shout_id}: {main_topic}")
# logger.debug(f"Final main_topic for shout#{shout_id}: {main_topic}")
if has_field(info, "authors") and hasattr(row, "authors"):
shout_dict["authors"] = (
json.loads(row.authors) if isinstance(row.authors, str) else row.authors
orjson.loads(row.authors) if isinstance(row.authors, str) else row.authors
)
if has_field(info, "media") and shout.media:
@@ -268,8 +267,8 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
media_data = shout.media
if isinstance(media_data, str):
try:
media_data = json.loads(media_data)
except json.JSONDecodeError:
media_data = orjson.loads(media_data)
except orjson.JSONDecodeError:
media_data = []
shout_dict["media"] = [media_data] if isinstance(media_data, dict) else media_data
@@ -413,26 +412,18 @@ async def load_shouts_search(_, info, text, options):
scores[shout_id] = sr.get("score")
hits_ids.append(shout_id)
""" q = (
q = (
query_with_stat(info)
if has_field(info, "stat")
else select(Shout).filter(and_(Shout.published_at.is_not(None), Shout.deleted_at.is_(None)))
) """
q = query_with_stat(info)
)
q = q.filter(Shout.id.in_(hits_ids))
q = apply_filters(q, options)
# added this to join topics
topic_join = aliased(ShoutTopic)
topic = aliased(Topic)
q = q.outerjoin(topic_join, topic_join.shout == Shout.id)
q = q.outerjoin(topic, topic.id == topic_join.topic)
q = apply_sorting(q, options)
shouts = get_shouts_with_links(info, q, limit, offset)
for shout in shouts:
shout["score"] = scores[f"{shout['id']}"]
shouts.sort(key=lambda x: x["score"], reverse=True)
shout.score = scores[f"{shout.id}"]
shouts.sort(key=lambda x: x.score, reverse=True)
return shouts
return []

View File

@@ -1,44 +1,240 @@
from sqlalchemy import select
from sqlalchemy import desc, select, text
from cache.cache import (
cache_topic,
cached_query,
get_cached_topic_authors,
get_cached_topic_by_slug,
get_cached_topic_followers,
invalidate_cache_by_prefix,
)
from cache.memorycache import cache_region
from orm.author import Author
from orm.topic import Topic
from resolvers.stat import get_with_stat
from services.auth import login_required
from services.db import local_session
from services.redis import redis
from services.schema import mutation, query
from utils.logger import root_logger as logger
# Вспомогательная функция для получения всех тем без статистики
async def get_all_topics():
"""
Получает все темы без статистики.
Используется для случаев, когда нужен полный список тем без дополнительной информации.
Returns:
list: Список всех тем без статистики
"""
cache_key = "topics:all:basic"
# Функция для получения всех тем из БД
async def fetch_all_topics():
logger.debug("Получаем список всех тем из БД и кешируем результат")
with local_session() as session:
# Запрос на получение базовой информации о темах
topics_query = select(Topic)
topics = session.execute(topics_query).scalars().all()
# Преобразуем темы в словари
return [topic.dict() for topic in topics]
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_all_topics)
# Вспомогательная функция для получения тем со статистикой с пагинацией
async def get_topics_with_stats(limit=100, offset=0, community_id=None, by=None):
"""
Получает темы со статистикой с пагинацией.
Args:
limit: Максимальное количество возвращаемых тем
offset: Смещение для пагинации
community_id: Опциональный ID сообщества для фильтрации
by: Опциональный параметр сортировки
Returns:
list: Список тем с их статистикой
"""
# Формируем ключ кеша с помощью универсальной функции
cache_key = f"topics:stats:limit={limit}:offset={offset}:community_id={community_id}"
# Функция для получения тем из БД
async def fetch_topics_with_stats():
logger.debug(f"Выполняем запрос на получение тем со статистикой: limit={limit}, offset={offset}")
with local_session() as session:
# Базовый запрос для получения тем
base_query = select(Topic)
# Добавляем фильтр по сообществу, если указан
if community_id:
base_query = base_query.where(Topic.community == community_id)
# Применяем сортировку на основе параметра by
if by:
if isinstance(by, dict):
# Обработка словаря параметров сортировки
for field, direction in by.items():
column = getattr(Topic, field, None)
if column:
if direction.lower() == "desc":
base_query = base_query.order_by(desc(column))
else:
base_query = base_query.order_by(column)
elif by == "popular":
# Сортировка по популярности (количеству публикаций)
# Примечание: это требует дополнительного запроса или подзапроса
base_query = base_query.order_by(
desc(Topic.id)
) # Временно, нужно заменить на proper implementation
else:
# По умолчанию сортируем по ID в обратном порядке
base_query = base_query.order_by(desc(Topic.id))
else:
# По умолчанию сортируем по ID в обратном порядке
base_query = base_query.order_by(desc(Topic.id))
# Применяем лимит и смещение
base_query = base_query.limit(limit).offset(offset)
# Получаем темы
topics = session.execute(base_query).scalars().all()
topic_ids = [topic.id for topic in topics]
if not topic_ids:
return []
# Запрос на получение статистики по публикациям для выбранных тем
shouts_stats_query = f"""
SELECT st.topic, COUNT(DISTINCT s.id) as shouts_count
FROM shout_topic st
JOIN shout s ON st.shout = s.id AND s.deleted_at IS NULL
WHERE st.topic IN ({",".join(map(str, topic_ids))})
GROUP BY st.topic
"""
shouts_stats = {row[0]: row[1] for row in session.execute(text(shouts_stats_query))}
# Запрос на получение статистики по подписчикам для выбранных тем
followers_stats_query = f"""
SELECT topic, COUNT(DISTINCT follower) as followers_count
FROM topic_followers
WHERE topic IN ({",".join(map(str, topic_ids))})
GROUP BY topic
"""
followers_stats = {row[0]: row[1] for row in session.execute(text(followers_stats_query))}
# Формируем результат с добавлением статистики
result = []
for topic in topics:
topic_dict = topic.dict()
topic_dict["stat"] = {
"shouts": shouts_stats.get(topic.id, 0),
"followers": followers_stats.get(topic.id, 0),
}
result.append(topic_dict)
# Кешируем каждую тему отдельно для использования в других функциях
await cache_topic(topic_dict)
return result
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_topics_with_stats)
# Функция для инвалидации кеша тем
async def invalidate_topics_cache(topic_id=None):
"""
Инвалидирует кеши тем при изменении данных.
Args:
topic_id: Опциональный ID темы для точечной инвалидации.
Если не указан, инвалидируются все кеши тем.
"""
if topic_id:
# Точечная инвалидация конкретной темы
logger.debug(f"Инвалидация кеша для темы #{topic_id}")
specific_keys = [
f"topic:id:{topic_id}",
f"topic:authors:{topic_id}",
f"topic:followers:{topic_id}",
f"topic_shouts_{topic_id}",
]
# Получаем slug темы, если есть
with local_session() as session:
topic = session.query(Topic).filter(Topic.id == topic_id).first()
if topic and topic.slug:
specific_keys.append(f"topic:slug:{topic.slug}")
# Удаляем конкретные ключи
for key in specific_keys:
try:
await redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
except Exception as e:
logger.error(f"Ошибка при удалении ключа {key}: {e}")
# Также ищем и удаляем ключи коллекций, содержащих данные об этой теме
collection_keys = await redis.execute("KEYS", "topics:stats:*")
if collection_keys:
await redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей тем")
else:
# Общая инвалидация всех кешей тем
logger.debug("Полная инвалидация кеша тем")
await invalidate_cache_by_prefix("topics")
# Запрос на получение всех тем
@query.field("get_topics_all")
def get_topics_all(_, _info):
cache_key = "get_topics_all" # Ключ для кеша
async def get_topics_all(_, _info):
"""
Получает список всех тем без статистики.
@cache_region.cache_on_arguments(cache_key)
def _get_topics_all():
topics_query = select(Topic)
return get_with_stat(topics_query) # Получение тем с учетом статистики
Returns:
list: Список всех тем
"""
return await get_all_topics()
return _get_topics_all()
# Запрос на получение тем с пагинацией и статистикой
@query.field("get_topics_paginated")
async def get_topics_paginated(_, _info, limit=100, offset=0, by=None):
"""
Получает список тем с пагинацией и статистикой.
Args:
limit: Максимальное количество возвращаемых тем
offset: Смещение для пагинации
by: Опциональные параметры сортировки
Returns:
list: Список тем с их статистикой
"""
return await get_topics_with_stats(limit, offset, None, by)
# Запрос на получение тем по сообществу
@query.field("get_topics_by_community")
def get_topics_by_community(_, _info, community_id: int):
cache_key = f"get_topics_by_community_{community_id}" # Ключ для кеша
async def get_topics_by_community(_, _info, community_id: int, limit=100, offset=0, by=None):
"""
Получает список тем, принадлежащих указанному сообществу с пагинацией и статистикой.
@cache_region.cache_on_arguments(cache_key)
def _get_topics_by_community():
topics_by_community_query = select(Topic).where(Topic.community == community_id)
return get_with_stat(topics_by_community_query)
Args:
community_id: ID сообщества
limit: Максимальное количество возвращаемых тем
offset: Смещение для пагинации
by: Опциональные параметры сортировки
return _get_topics_by_community()
Returns:
list: Список тем с их статистикой
"""
return await get_topics_with_stats(limit, offset, community_id, by)
# Запрос на получение тем по автору
@@ -74,6 +270,9 @@ async def create_topic(_, _info, topic_input):
session.add(new_topic)
session.commit()
# Инвалидируем кеш всех тем
await invalidate_topics_cache()
return {"topic": new_topic}
@@ -87,10 +286,19 @@ async def update_topic(_, _info, topic_input):
if not topic:
return {"error": "topic not found"}
else:
old_slug = topic.slug
Topic.update(topic, topic_input)
session.add(topic)
session.commit()
# Инвалидируем кеш только для этой конкретной темы
await invalidate_topics_cache(topic.id)
# Если slug изменился, удаляем старый ключ
if old_slug != topic.slug:
await redis.execute("DEL", f"topic:slug:{old_slug}")
logger.debug(f"Удален ключ кеша для старого slug: {old_slug}")
return {"topic": topic}
@@ -111,6 +319,11 @@ async def delete_topic(_, info, slug: str):
session.delete(t)
session.commit()
# Инвалидируем кеш всех тем и конкретной темы
await invalidate_topics_cache()
await redis.execute("DEL", f"topic:slug:{slug}")
await redis.execute("DEL", f"topic:id:{t.id}")
return {}
return {"error": "access denied"}

View File

@@ -26,6 +26,9 @@ type Query {
load_shout_ratings(shout: Int!, limit: Int, offset: Int): [Reaction]
load_comment_ratings(comment: Int!, limit: Int, offset: Int): [Reaction]
# branched comments pagination
load_comments_branch(shout: Int!, parent_id: Int, limit: Int, offset: Int, sort: ReactionSort, children_limit: Int, children_offset: Int): [Reaction]
# reader
get_shout(slug: String, shout_id: Int): Shout
load_shouts_by(options: LoadShoutsOptions): [Shout]

View File

@@ -55,6 +55,7 @@ type Reaction {
stat: Stat
oid: String
# old_thread: String
first_replies: [Reaction]
}
type MediaItem {
@@ -207,7 +208,6 @@ type CommonResult {
}
type SearchResult {
id: Int!
slug: String!
title: String!
cover: String

View File

@@ -1,34 +0,0 @@
import sys
from pathlib import Path
from granian.constants import Interfaces
from granian.log import LogLevels
from granian.server import Server
from settings import PORT
from utils.logger import root_logger as logger
if __name__ == "__main__":
logger.info("started")
try:
granian_instance = Server(
"main:app",
address="0.0.0.0",
port=PORT,
interface=Interfaces.ASGI,
workers=1,
websockets=False,
log_level=LogLevels.debug,
backlog=2048,
)
if "dev" in sys.argv:
logger.info("dev mode, building ssl context")
granian_instance.build_ssl_context(cert=Path("localhost.pem"), key=Path("localhost-key.pem"), password=None)
granian_instance.serve()
except Exception as error:
logger.error(error, exc_info=True)
raise
finally:
logger.info("stopped")

View File

@@ -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:
@@ -181,27 +259,3 @@ def get_json_builder():
# Используем их в коде
json_builder, json_array_builder, json_cast = get_json_builder()
async def fetch_all_shouts(session=None):
"""Fetch all published shouts for search indexing"""
from orm.shout import Shout
close_session = False
if session is None:
session = local_session()
close_session = True
try:
# Fetch only published and non-deleted shouts
query = session.query(Shout).filter(
Shout.published_at.is_not(None),
Shout.deleted_at.is_(None)
)
shouts = query.all()
return shouts
except Exception as e:
logger.error(f"Error fetching shouts for search indexing: {e}")
return []
finally:
if close_session:
session.close()

View File

@@ -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:

View File

@@ -2,571 +2,231 @@ import asyncio
import json
import logging
import os
import httpx
import time
import random
# Set up proper logging
import orjson
from opensearchpy import OpenSearch
from services.redis import redis
from utils.encoders import CustomJSONEncoder
# Set redis logging level to suppress DEBUG messages
logger = logging.getLogger("search")
logger.setLevel(logging.INFO) # Change to INFO to see more details
logger.setLevel(logging.WARNING)
# Configuration for search service
SEARCH_ENABLED = bool(os.environ.get("SEARCH_ENABLED", "true").lower() in ["true", "1", "yes"])
TXTAI_SERVICE_URL = os.environ.get("TXTAI_SERVICE_URL", "none")
MAX_BATCH_SIZE = int(os.environ.get("SEARCH_MAX_BATCH_SIZE", "25"))
ELASTIC_HOST = os.environ.get("ELASTIC_HOST", "").replace("https://", "")
ELASTIC_USER = os.environ.get("ELASTIC_USER", "")
ELASTIC_PASSWORD = os.environ.get("ELASTIC_PASSWORD", "")
ELASTIC_PORT = os.environ.get("ELASTIC_PORT", 9200)
ELASTIC_URL = os.environ.get(
"ELASTIC_URL",
f"https://{ELASTIC_USER}:{ELASTIC_PASSWORD}@{ELASTIC_HOST}:{ELASTIC_PORT}",
)
REDIS_TTL = 86400 # 1 день в секундах
index_settings = {
"settings": {
"index": {"number_of_shards": 1, "auto_expand_replicas": "0-all"},
"analysis": {
"analyzer": {
"ru": {
"tokenizer": "standard",
"filter": ["lowercase", "ru_stop", "ru_stemmer"],
}
},
"filter": {
"ru_stemmer": {"type": "stemmer", "language": "russian"},
"ru_stop": {"type": "stop", "stopwords": "_russian_"},
},
},
},
"mappings": {
"properties": {
"body": {"type": "text", "analyzer": "ru"},
"title": {"type": "text", "analyzer": "ru"},
"subtitle": {"type": "text", "analyzer": "ru"},
"lead": {"type": "text", "analyzer": "ru"},
"media": {"type": "text", "analyzer": "ru"},
}
},
}
expected_mapping = index_settings["mappings"]
# Создание цикла событий
search_loop = asyncio.get_event_loop()
# В начале файла добавим флаг
SEARCH_ENABLED = bool(os.environ.get("ELASTIC_HOST", ""))
def get_indices_stats():
indices_stats = search_service.client.cat.indices(format="json")
for index_info in indices_stats:
index_name = index_info["index"]
if not index_name.startswith("."):
index_health = index_info["health"]
index_status = index_info["status"]
pri_shards = index_info["pri"]
rep_shards = index_info["rep"]
docs_count = index_info["docs.count"]
docs_deleted = index_info["docs.deleted"]
store_size = index_info["store.size"]
pri_store_size = index_info["pri.store.size"]
logger.info(f"Index: {index_name}")
logger.info(f"Health: {index_health}")
logger.info(f"Status: {index_status}")
logger.info(f"Primary Shards: {pri_shards}")
logger.info(f"Replica Shards: {rep_shards}")
logger.info(f"Documents Count: {docs_count}")
logger.info(f"Deleted Documents: {docs_deleted}")
logger.info(f"Store Size: {store_size}")
logger.info(f"Primary Store Size: {pri_store_size}")
class SearchService:
def __init__(self):
logger.info(f"Initializing search service with URL: {TXTAI_SERVICE_URL}")
self.available = SEARCH_ENABLED
# Use different timeout settings for indexing and search requests
self.client = httpx.AsyncClient(timeout=30.0, base_url=TXTAI_SERVICE_URL)
self.index_client = httpx.AsyncClient(timeout=120.0, base_url=TXTAI_SERVICE_URL)
def __init__(self, index_name="search_index"):
logger.info("Инициализируем поиск...")
self.index_name = index_name
self.client = None
self.lock = asyncio.Lock()
# Инициализация клиента OpenSearch только если поиск включен
if SEARCH_ENABLED:
try:
self.client = OpenSearch(
hosts=[{"host": ELASTIC_HOST, "port": ELASTIC_PORT}],
http_compress=True,
http_auth=(ELASTIC_USER, ELASTIC_PASSWORD),
use_ssl=True,
verify_certs=False,
ssl_assert_hostname=False,
ssl_show_warn=False,
)
logger.info("Клиент OpenSearch.org подключен")
search_loop.create_task(self.check_index())
except Exception as exc:
logger.warning(f"Поиск отключен из-за ошибки подключения: {exc}")
self.client = None
else:
logger.info("Поиск отключен (ELASTIC_HOST не установлен)")
if not self.available:
logger.info("Search disabled (SEARCH_ENABLED = False)")
async def info(self):
"""Return information about search service"""
if not self.available:
if not SEARCH_ENABLED:
return {"status": "disabled"}
try:
response = await self.client.get("/info")
response.raise_for_status()
result = response.json()
logger.info(f"Search service info: {result}")
return result
return get_indices_stats()
except Exception as e:
logger.error(f"Failed to get search info: {e}")
return {"status": "error", "message": str(e)}
def is_ready(self):
"""Check if service is available"""
return self.available
async def verify_docs(self, doc_ids):
"""Verify which documents exist in the search index"""
if not self.available:
return {"status": "disabled"}
try:
logger.info(f"Verifying {len(doc_ids)} documents in search index")
response = await self.client.post(
"/verify-docs",
json={"doc_ids": doc_ids},
timeout=60.0 # Longer timeout for potentially large ID lists
)
response.raise_for_status()
result = response.json()
# Log summary of verification results
missing_count = len(result.get("missing", []))
logger.info(f"Document verification complete: {missing_count} missing out of {len(doc_ids)} total")
return result
except Exception as e:
logger.error(f"Document verification error: {e}")
return {"status": "error", "message": str(e)}
def delete_index(self):
if self.client:
logger.warning(f"[!!!] Удаляем индекс {self.index_name}")
self.client.indices.delete(index=self.index_name, ignore_unavailable=True)
def create_index(self):
if self.client:
logger.info(f"Создается индекс: {self.index_name}")
self.client.indices.create(index=self.index_name, body=index_settings)
logger.info(f"Индекс {self.index_name} создан")
async def check_index(self):
if self.client:
logger.info(f"Проверяем индекс {self.index_name}...")
if not self.client.indices.exists(index=self.index_name):
self.create_index()
self.client.indices.put_mapping(index=self.index_name, body=expected_mapping)
else:
logger.info(f"Найден существующий индекс {self.index_name}")
# Проверка и обновление структуры индекса, если необходимо
result = self.client.indices.get_mapping(index=self.index_name)
if isinstance(result, str):
result = orjson.loads(result)
if isinstance(result, dict):
mapping = result.get(self.index_name, {}).get("mappings")
logger.info(f"Найдена структура индексации: {mapping['properties'].keys()}")
expected_keys = expected_mapping["properties"].keys()
if mapping and mapping["properties"].keys() != expected_keys:
logger.info(f"Ожидаемая структура индексации: {expected_mapping}")
logger.warning("[!!!] Требуется переиндексация всех данных")
self.delete_index()
self.client = None
else:
logger.error("клиент не инициализован, невозможно проверить индекс")
def index(self, shout):
"""Index a single document"""
if not self.available:
if not SEARCH_ENABLED:
return
logger.info(f"Indexing post {shout.id}")
# Start in background to not block
asyncio.create_task(self.perform_index(shout))
if self.client:
logger.info(f"Индексируем пост {shout.id}")
index_body = {
"body": shout.body,
"title": shout.title,
"subtitle": shout.subtitle,
"lead": shout.lead,
"media": shout.media,
}
asyncio.create_task(self.perform_index(shout, index_body))
async def perform_index(self, shout):
"""Actually perform the indexing operation"""
if not self.available:
return
try:
# Combine all text fields
text = " ".join(filter(None, [
shout.title or "",
shout.subtitle or "",
shout.lead or "",
shout.body or "",
shout.media or ""
]))
if not text.strip():
logger.warning(f"No text content to index for shout {shout.id}")
return
logger.info(f"Indexing document: ID={shout.id}, Text length={len(text)}")
# Send to txtai service
response = await self.client.post(
"/index",
json={"id": str(shout.id), "text": text}
)
response.raise_for_status()
result = response.json()
logger.info(f"Post {shout.id} successfully indexed: {result}")
except Exception as e:
logger.error(f"Indexing error for shout {shout.id}: {e}")
async def bulk_index(self, shouts):
"""Index multiple documents at once with adaptive batch sizing"""
if not self.available or not shouts:
logger.warning(f"Bulk indexing skipped: available={self.available}, shouts_count={len(shouts) if shouts else 0}")
return
start_time = time.time()
logger.info(f"Starting bulk indexing of {len(shouts)} documents")
MAX_TEXT_LENGTH = 4000 # Maximum text length to send in a single request
max_batch_size = MAX_BATCH_SIZE
total_indexed = 0
total_skipped = 0
total_truncated = 0
total_retries = 0
# Group documents by size to process smaller documents in larger batches
small_docs = []
medium_docs = []
large_docs = []
# First pass: prepare all documents and categorize by size
for shout in shouts:
async def perform_index(self, shout, index_body):
if self.client:
try:
text_fields = []
for field_name in ['title', 'subtitle', 'lead', 'body']:
field_value = getattr(shout, field_name, None)
if field_value and isinstance(field_value, str) and field_value.strip():
text_fields.append(field_value.strip())
# Media field processing remains the same
media = getattr(shout, 'media', None)
if media:
# Your existing media processing logic
if isinstance(media, str):
try:
media_json = json.loads(media)
if isinstance(media_json, dict):
if 'title' in media_json:
text_fields.append(media_json['title'])
if 'body' in media_json:
text_fields.append(media_json['body'])
except json.JSONDecodeError:
text_fields.append(media)
elif isinstance(media, dict):
if 'title' in media:
text_fields.append(media['title'])
if 'body' in media:
text_fields.append(media['body'])
text = " ".join(text_fields)
if not text.strip():
logger.debug(f"Skipping shout {shout.id}: no text content")
total_skipped += 1
continue
# Truncate text if it exceeds the maximum length
original_length = len(text)
if original_length > MAX_TEXT_LENGTH:
text = text[:MAX_TEXT_LENGTH]
logger.info(f"Truncated document {shout.id} from {original_length} to {MAX_TEXT_LENGTH} chars")
total_truncated += 1
document = {
"id": str(shout.id),
"text": text
}
# Categorize by size
text_len = len(text)
if text_len > 5000:
large_docs.append(document)
elif text_len > 2000:
medium_docs.append(document)
else:
small_docs.append(document)
total_indexed += 1
except Exception as e:
logger.error(f"Error processing shout {getattr(shout, 'id', 'unknown')} for indexing: {e}")
total_skipped += 1
# Process each category with appropriate batch sizes
logger.info(f"Documents categorized: {len(small_docs)} small, {len(medium_docs)} medium, {len(large_docs)} large")
# Process small documents (larger batches)
if small_docs:
batch_size = min(max_batch_size, 15)
await self._process_document_batches(small_docs, batch_size, "small")
# Process medium documents (medium batches)
if medium_docs:
batch_size = min(max_batch_size, 10)
await self._process_document_batches(medium_docs, batch_size, "medium")
# Process large documents (small batches)
if large_docs:
batch_size = min(max_batch_size, 3)
await self._process_document_batches(large_docs, batch_size, "large")
elapsed = time.time() - start_time
logger.info(f"Bulk indexing completed in {elapsed:.2f}s: {total_indexed} indexed, {total_skipped} skipped, {total_truncated} truncated, {total_retries} retries")
async def _process_document_batches(self, documents, batch_size, size_category):
"""Process document batches with retry logic"""
# Check for possible database corruption before starting
db_error_count = 0
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
batch_id = f"{size_category}-{i//batch_size + 1}"
logger.info(f"Processing {size_category} batch {batch_id} of {len(batch)} documents")
retry_count = 0
max_retries = 3
success = False
# Process with retries
while not success and retry_count < max_retries:
try:
if batch:
sample = batch[0]
logger.info(f"Sample document in batch {batch_id}: id={sample['id']}, text_length={len(sample['text'])}")
logger.info(f"Sending batch {batch_id} of {len(batch)} documents to search service (attempt {retry_count+1})")
response = await self.index_client.post(
"/bulk-index",
json=batch,
timeout=120.0 # Explicit longer timeout for large batches
)
# Handle 422 validation errors - these won't be fixed by retrying
if response.status_code == 422:
error_detail = response.json()
truncated_error = self._truncate_error_detail(error_detail)
logger.error(f"Validation error from search service for batch {batch_id}: {truncated_error}")
break
# Handle 500 server errors - these might be fixed by retrying with smaller batches
elif response.status_code == 500:
db_error_count += 1
# If we've seen multiple 500s, log a critical error
if db_error_count >= 3:
logger.critical(f"Multiple server errors detected (500). The search service may need manual intervention. Stopping batch {batch_id} processing.")
break
# Try again with exponential backoff
if retry_count < max_retries - 1:
retry_count += 1
wait_time = (2 ** retry_count) + (random.random() * 0.5) # Exponential backoff with jitter
logger.warning(f"Server error for batch {batch_id}, retrying in {wait_time:.1f}s (attempt {retry_count+1}/{max_retries})")
await asyncio.sleep(wait_time)
continue
# Final retry, split the batch
elif len(batch) > 1:
logger.warning(f"Splitting batch {batch_id} after repeated failures")
mid = len(batch) // 2
await self._process_single_batch(batch[:mid], f"{batch_id}-A")
await self._process_single_batch(batch[mid:], f"{batch_id}-B")
break
else:
# Can't split a single document
logger.error(f"Failed to index document {batch[0]['id']} after {max_retries} attempts")
break
# Normal success case
response.raise_for_status()
result = response.json()
logger.info(f"Batch {batch_id} indexed successfully: {result}")
success = True
db_error_count = 0 # Reset error counter on success
except Exception as e:
# Check if it looks like a database corruption error
error_str = str(e).lower()
if "duplicate key" in error_str or "unique constraint" in error_str or "nonetype" in error_str:
db_error_count += 1
if db_error_count >= 2:
logger.critical(f"Potential database corruption detected: {error_str}. The search service may need manual intervention. Stopping batch {batch_id} processing.")
break
if retry_count < max_retries - 1:
retry_count += 1
wait_time = (2 ** retry_count) + (random.random() * 0.5)
logger.warning(f"Error for batch {batch_id}, retrying in {wait_time:.1f}s: {str(e)[:200]}")
await asyncio.sleep(wait_time)
else:
# Last resort - try to split the batch
if len(batch) > 1:
logger.warning(f"Splitting batch {batch_id} after exception: {str(e)[:200]}")
mid = len(batch) // 2
await self._process_single_batch(batch[:mid], f"{batch_id}-A")
await self._process_single_batch(batch[mid:], f"{batch_id}-B")
else:
logger.error(f"Failed to index document {batch[0]['id']} after {max_retries} attempts: {e}")
break
async def _process_single_batch(self, documents, batch_id):
"""Process a single batch with maximum reliability"""
max_retries = 3
retry_count = 0
while retry_count < max_retries:
try:
if not documents:
return
logger.info(f"Processing sub-batch {batch_id} with {len(documents)} documents")
response = await self.index_client.post(
"/bulk-index",
json=documents,
timeout=90.0
await asyncio.wait_for(
self.client.index(index=self.index_name, id=str(shout.id), body=index_body), timeout=40.0
)
response.raise_for_status()
result = response.json()
logger.info(f"Sub-batch {batch_id} indexed successfully: {result}")
return # Success, exit the retry loop
except asyncio.TimeoutError:
logger.error(f"Indexing timeout for shout {shout.id}")
except Exception as e:
error_str = str(e).lower()
retry_count += 1
# Check if it's a transient error that txtai might recover from internally
if "dictionary changed size" in error_str or "transaction error" in error_str:
wait_time = (2 ** retry_count) + (random.random() * 0.5)
logger.warning(f"Transient txtai error in sub-batch {batch_id}, waiting {wait_time:.1f}s for recovery: {str(e)[:200]}")
await asyncio.sleep(wait_time) # Wait for txtai to recover
continue # Try again
# For other errors or final retry failure
logger.error(f"Error indexing sub-batch {batch_id} (attempt {retry_count}/{max_retries}): {str(e)[:200]}")
# Only try one-by-one on the final retry
if retry_count >= max_retries and len(documents) > 1:
logger.info(f"Processing documents in sub-batch {batch_id} individually")
for i, doc in enumerate(documents):
try:
resp = await self.index_client.post("/index", json=doc, timeout=30.0)
resp.raise_for_status()
logger.info(f"Indexed document {doc['id']} individually")
except Exception as e2:
logger.error(f"Failed to index document {doc['id']} individually: {str(e2)[:100]}")
return # Exit after individual processing attempt
def _truncate_error_detail(self, error_detail):
"""Truncate error details for logging"""
truncated_detail = error_detail.copy() if isinstance(error_detail, dict) else error_detail
if isinstance(truncated_detail, dict) and 'detail' in truncated_detail and isinstance(truncated_detail['detail'], list):
for i, item in enumerate(truncated_detail['detail']):
if isinstance(item, dict) and 'input' in item:
if isinstance(item['input'], dict) and any(k in item['input'] for k in ['documents', 'text']):
# Check for documents list
if 'documents' in item['input'] and isinstance(item['input']['documents'], list):
for j, doc in enumerate(item['input']['documents']):
if 'text' in doc and isinstance(doc['text'], str) and len(doc['text']) > 100:
item['input']['documents'][j]['text'] = f"{doc['text'][:100]}... [truncated, total {len(doc['text'])} chars]"
# Check for direct text field
if 'text' in item['input'] and isinstance(item['input']['text'], str) and len(item['input']['text']) > 100:
item['input']['text'] = f"{item['input']['text'][:100]}... [truncated, total {len(item['input']['text'])} chars]"
return truncated_detail
logger.error(f"Indexing error for shout {shout.id}: {e}")
async def search(self, text, limit, offset):
"""Search documents"""
if not self.available:
logger.warning("Search not available")
if not SEARCH_ENABLED:
return []
if not isinstance(text, str) or not text.strip():
logger.warning(f"Invalid search text: {text}")
return []
logger.info(f"Searching for: '{text}' (limit={limit}, offset={offset})")
try:
logger.info(f"Sending search request: text='{text}', limit={limit}, offset={offset}")
response = await self.client.post(
"/search",
json={"text": text, "limit": limit, "offset": offset}
logger.info(f"Ищем: {text} {offset}+{limit}")
search_body = {
"query": {"multi_match": {"query": text, "fields": ["title", "lead", "subtitle", "body", "media"]}}
}
if self.client:
search_response = self.client.search(
index=self.index_name,
body=search_body,
size=limit,
from_=offset,
_source=False,
_source_excludes=["title", "body", "subtitle", "media", "lead", "_index"],
)
response.raise_for_status()
logger.info(f"Raw search response: {response.text}")
result = response.json()
logger.info(f"Parsed search response: {result}")
formatted_results = result.get("results", [])
logger.info(f"Search for '{text}' returned {len(formatted_results)} results")
if formatted_results:
logger.info(f"Sample result: {formatted_results[0]}")
else:
logger.warning(f"No results found for '{text}'")
return formatted_results
except Exception as e:
logger.error(f"Search error for '{text}': {e}", exc_info=True)
return []
async def check_index_status(self):
"""Get detailed statistics about the search index health"""
if not self.available:
return {"status": "disabled"}
try:
response = await self.client.get("/index-status")
response.raise_for_status()
result = response.json()
logger.info(f"Index status check: {result['status']}, {result['documents_count']} documents")
# Log warnings for any inconsistencies
if result.get("consistency", {}).get("status") != "ok":
null_count = result.get("consistency", {}).get("null_embeddings_count", 0)
if null_count > 0:
logger.warning(f"Found {null_count} documents with NULL embeddings")
return result
except Exception as e:
logger.error(f"Failed to check index status: {e}")
return {"status": "error", "message": str(e)}
hits = search_response["hits"]["hits"]
results = [{"id": hit["_id"], "score": hit["_score"]} for hit in hits]
# если результаты не пустые
if results:
# Кэширование в Redis с TTL
redis_key = f"search:{text}:{offset}+{limit}"
await redis.execute(
"SETEX",
redis_key,
REDIS_TTL,
json.dumps(results, cls=CustomJSONEncoder),
)
return results
return []
# Create the search service singleton
search_service = SearchService()
# API-compatible function to perform a search
async def search_text(text: str, limit: int = 50, offset: int = 0):
payload = []
if search_service.available:
if search_service.client:
# Использование метода search_post из OpenSearchService
payload = await search_service.search(text, limit, offset)
return payload
async def initialize_search_index(shouts_data):
"""Initialize search index with existing data during application startup"""
if not SEARCH_ENABLED:
logger.info("Search indexing skipped (SEARCH_ENABLED=False)")
return
if not shouts_data:
logger.warning("No shouts data provided for search indexing")
return
logger.info(f"Checking search index status for {len(shouts_data)} documents")
# Get the current index info
info = await search_service.info()
if info.get("status") in ["error", "unavailable", "disabled"]:
logger.error(f"Cannot initialize search index: {info}")
return
# Check if index has approximately right number of documents
index_stats = info.get("index_stats", {})
indexed_doc_count = index_stats.get("document_count", 0)
# Add a more detailed status check
index_status = await search_service.check_index_status()
if index_status.get("status") == "healthy":
logger.info("Index status check passed")
elif index_status.get("status") == "inconsistent":
logger.warning("Index status check found inconsistencies")
# Get documents with null embeddings
problem_ids = index_status.get("consistency", {}).get("null_embeddings_sample", [])
if problem_ids:
logger.info(f"Repairing {len(problem_ids)} documents with NULL embeddings")
problem_docs = [shout for shout in shouts_data if str(shout.id) in problem_ids]
if problem_docs:
await search_service.bulk_index(problem_docs)
# Log database document summary
db_ids = [str(shout.id) for shout in shouts_data]
logger.info(f"Database contains {len(shouts_data)} documents. Sample IDs: {', '.join(db_ids[:5])}...")
# Calculate summary by ID range to understand the coverage
try:
# Parse numeric IDs where possible to analyze coverage
numeric_ids = [int(sid) for sid in db_ids if sid.isdigit()]
if numeric_ids:
min_id = min(numeric_ids)
max_id = max(numeric_ids)
id_range = max_id - min_id + 1
coverage_pct = (len(numeric_ids) / id_range) * 100 if id_range > 0 else 0
logger.info(f"ID range analysis: min_id={min_id}, max_id={max_id}, range={id_range}, "
f"coverage={coverage_pct:.1f}% ({len(numeric_ids)}/{id_range})")
except Exception as e:
logger.warning(f"Could not analyze ID ranges: {e}")
# If counts are significantly different, do verification
if abs(indexed_doc_count - len(shouts_data)) > 10:
logger.info(f"Document count mismatch: {indexed_doc_count} in index vs {len(shouts_data)} in database. Verifying...")
# Get all document IDs from your database
doc_ids = [str(shout.id) for shout in shouts_data]
# Verify which ones are missing from the index
verification = await search_service.verify_docs(doc_ids)
if verification.get("status") == "error":
logger.error(f"Document verification failed: {verification.get('message')}")
return
# Index only missing documents
missing_ids = verification.get("missing", [])
if missing_ids:
logger.info(f"Found {len(missing_ids)} documents missing from index. Indexing them...")
logger.info(f"Sample missing IDs: {', '.join(missing_ids[:10])}...")
missing_docs = [shout for shout in shouts_data if str(shout.id) in missing_ids]
await search_service.bulk_index(missing_docs)
else:
logger.info("All documents are already indexed.")
else:
logger.info(f"Search index appears to be in sync ({indexed_doc_count} documents indexed).")
# Optional sample verification (can be slow with large document sets)
# Uncomment if you want to periodically check a random sample even when counts match
"""
sample_size = 10
if len(db_ids) > sample_size:
sample_ids = random.sample(db_ids, sample_size)
logger.info(f"Performing random sample verification on {sample_size} documents...")
verification = await search_service.verify_docs(sample_ids)
if verification.get("missing"):
missing_count = len(verification.get("missing", []))
logger.warning(f"Random verification found {missing_count}/{sample_size} missing docs "
f"despite count match. Consider full verification.")
else:
logger.info("Random document sample verification passed.")
"""
# Verify with test query
try:
test_query = "test"
logger.info(f"Verifying search index with query: '{test_query}'")
test_results = await search_text(test_query, 5)
if test_results:
logger.info(f"Search verification successful: found {len(test_results)} results")
# Log categories covered by search results
categories = set()
for result in test_results:
result_id = result.get("id")
matching_shouts = [s for s in shouts_data if str(s.id) == result_id]
if matching_shouts and hasattr(matching_shouts[0], 'category'):
categories.add(getattr(matching_shouts[0], 'category', 'unknown'))
if categories:
logger.info(f"Search results cover categories: {', '.join(categories)}")
else:
logger.warning("Search verification returned no results. Index may be empty or not working.")
except Exception as e:
logger.error(f"Error verifying search index: {e}")
# Проверить что URL корректный
OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "rc1a-3n5pi3bhuj9gieel.mdb.yandexcloud.net")

View File

@@ -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.")

View File

@@ -1,18 +1,24 @@
import sys
from os import environ
PORT = 8000
MODE = "development" if "dev" in sys.argv else "production"
DEV_SERVER_PID_FILE_NAME = "dev-server.pid"
PORT = environ.get("PORT") or 8000
# storages
DB_URL = (
environ.get("DATABASE_URL", "").replace("postgres://", "postgresql://")
or environ.get("DB_URL", "").replace("postgres://", "postgresql://")
or "sqlite:///discoursio.db"
)
REDIS_URL = environ.get("REDIS_URL") or "redis://127.0.0.1"
AUTH_URL = environ.get("AUTH_URL") or ""
GLITCHTIP_DSN = environ.get("GLITCHTIP_DSN")
DEV_SERVER_PID_FILE_NAME = "dev-server.pid"
MODE = "development" if "dev" in sys.argv else "production"
# debug
GLITCHTIP_DSN = environ.get("GLITCHTIP_DSN")
# authorizer.dev
AUTH_URL = environ.get("AUTH_URL") or "https://auth.discours.io/graphql"
ADMIN_SECRET = environ.get("AUTH_SECRET") or "nothing"
WEBHOOK_SECRET = environ.get("WEBHOOK_SECRET") or "nothing-else"

View File

@@ -1,9 +1,28 @@
import json
from decimal import Decimal
from json import JSONEncoder
class CustomJSONEncoder(json.JSONEncoder):
class CustomJSONEncoder(JSONEncoder):
"""
Расширенный JSON энкодер с поддержкой сериализации объектов SQLAlchemy.
Примеры:
>>> import json
>>> from decimal import Decimal
>>> from orm.topic import Topic
>>> json.dumps(Decimal("10.50"), cls=CustomJSONEncoder)
'"10.50"'
>>> topic = Topic(id=1, slug="test")
>>> json.dumps(topic, cls=CustomJSONEncoder)
'{"id": 1, "slug": "test", ...}'
"""
def default(self, obj):
if isinstance(obj, Decimal):
return str(obj)
# Проверяем, есть ли у объекта метод dict() (как у моделей SQLAlchemy)
if hasattr(obj, "dict") and callable(obj.dict):
return obj.dict()
return super().default(obj)

View File

@@ -13,7 +13,7 @@ def filter(record: logging.LogRecord):
record.emoji = (
"🔍"
if record.levelno == logging.DEBUG
else "🖊️"
else ""
if record.levelno == logging.INFO
else "🚧"
if record.levelno == logging.WARNING