This commit is contained in:
parent
74e639737e
commit
67fa44b062
|
@ -22,6 +22,7 @@ T = TypeVar("T")
|
||||||
REGISTRY: Dict[str, type] = {}
|
REGISTRY: Dict[str, type] = {}
|
||||||
Base = declarative_base()
|
Base = declarative_base()
|
||||||
|
|
||||||
|
|
||||||
def profile_sqlalchemy_queries(threshold=0.1):
|
def profile_sqlalchemy_queries(threshold=0.1):
|
||||||
def decorator(fn):
|
def decorator(fn):
|
||||||
@wraps(fn)
|
@wraps(fn)
|
||||||
|
@ -30,12 +31,15 @@ def profile_sqlalchemy_queries(threshold=0.1):
|
||||||
if elapsed is not None:
|
if elapsed is not None:
|
||||||
print(f"Query took {elapsed:.3f} seconds to execute.")
|
print(f"Query took {elapsed:.3f} seconds to execute.")
|
||||||
stats = stat_loader()
|
stats = stat_loader()
|
||||||
stats.sort_stats('cumulative')
|
stats.sort_stats("cumulative")
|
||||||
stats.print_stats()
|
stats.print_stats()
|
||||||
return result
|
return result
|
||||||
|
|
||||||
return wrapper
|
return wrapper
|
||||||
|
|
||||||
return decorator
|
return decorator
|
||||||
|
|
||||||
|
|
||||||
def _profile(fn, threshold, *args, **kw):
|
def _profile(fn, threshold, *args, **kw):
|
||||||
began = time.time()
|
began = time.time()
|
||||||
result = fn(*args, **kw)
|
result = fn(*args, **kw)
|
||||||
|
@ -46,24 +50,27 @@ def _profile(fn, threshold, *args, **kw):
|
||||||
else:
|
else:
|
||||||
return None, None, result
|
return None, None, result
|
||||||
|
|
||||||
|
|
||||||
# Перехватчики для журнала запросов SQLAlchemy
|
# Перехватчики для журнала запросов SQLAlchemy
|
||||||
@event.listens_for(Engine, "before_cursor_execute")
|
@event.listens_for(Engine, "before_cursor_execute")
|
||||||
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
|
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
|
||||||
conn._query_start_time = time.time()
|
conn._query_start_time = time.time()
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(Engine, "after_cursor_execute")
|
@event.listens_for(Engine, "after_cursor_execute")
|
||||||
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
|
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
|
||||||
if hasattr(conn, '_query_start_time'):
|
if hasattr(conn, "_query_start_time"):
|
||||||
elapsed = time.time() - conn._query_start_time
|
elapsed = time.time() - conn._query_start_time
|
||||||
del conn._query_start_time
|
del conn._query_start_time
|
||||||
if elapsed > 0.2: # Adjust threshold as needed
|
if elapsed > 0.2: # Adjust threshold as needed
|
||||||
logger.debug(f"{'*' * math.floor(elapsed)} {elapsed:.3f} seconds to execute.")
|
logger.debug(
|
||||||
|
f"{'*' * math.floor(elapsed)} {elapsed:.3f} seconds to execute."
|
||||||
|
)
|
||||||
# Profile the query if execution time exceeds the threshold
|
# Profile the query if execution time exceeds the threshold
|
||||||
profiler = profile_sqlalchemy_queries(threshold=0.2)(cursor.execute)
|
profiler = profile_sqlalchemy_queries(threshold=0.2)(cursor.execute)
|
||||||
profiler(statement, parameters)
|
profiler(statement, parameters)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def local_session(src=""):
|
def local_session(src=""):
|
||||||
return Session(bind=engine, expire_on_commit=False)
|
return Session(bind=engine, expire_on_commit=False)
|
||||||
|
|
||||||
|
|
|
@ -14,34 +14,34 @@ from services.viewed import ViewedStorage
|
||||||
|
|
||||||
@event.listens_for(Author, "after_insert")
|
@event.listens_for(Author, "after_insert")
|
||||||
@event.listens_for(Author, "after_update")
|
@event.listens_for(Author, "after_update")
|
||||||
def after_author_update(mapper, connection, target):
|
def after_author_update(mapper, connection, target: Author):
|
||||||
redis_key = f"user:{target.user}:author"
|
redis_key = f"user:{target.user}:author"
|
||||||
asyncio.create_task(redis.execute("set", redis_key, json.dumps(vars(target))))
|
asyncio.create_task(redis.execute("set", redis_key, json.dumps(target.dict())))
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(TopicFollower, "after_insert")
|
@event.listens_for(TopicFollower, "after_insert")
|
||||||
def after_topic_follower_insert(mapper, connection, target):
|
def after_topic_follower_insert(mapper, connection, target: TopicFollower):
|
||||||
asyncio.create_task(
|
asyncio.create_task(
|
||||||
handle_topic_follower_change(connection, target.topic, target.follower, True)
|
handle_topic_follower_change(connection, target.topic, target.follower, True)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(TopicFollower, "after_delete")
|
@event.listens_for(TopicFollower, "after_delete")
|
||||||
def after_topic_follower_delete(mapper, connection, target):
|
def after_topic_follower_delete(mapper, connection, target: TopicFollower):
|
||||||
asyncio.create_task(
|
asyncio.create_task(
|
||||||
handle_topic_follower_change(connection, target.topic, target.follower, False)
|
handle_topic_follower_change(connection, target.topic, target.follower, False)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(AuthorFollower, "after_insert")
|
@event.listens_for(AuthorFollower, "after_insert")
|
||||||
def after_author_follower_insert(mapper, connection, target):
|
def after_author_follower_insert(mapper, connection, target: AuthorFollower):
|
||||||
asyncio.create_task(
|
asyncio.create_task(
|
||||||
handle_author_follower_change(connection, target.author, target.follower, True)
|
handle_author_follower_change(connection, target.author, target.follower, True)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(AuthorFollower, "after_delete")
|
@event.listens_for(AuthorFollower, "after_delete")
|
||||||
def after_author_follower_delete(mapper, connection, target):
|
def after_author_follower_delete(mapper, connection, target: AuthorFollower):
|
||||||
asyncio.create_task(
|
asyncio.create_task(
|
||||||
handle_author_follower_change(connection, target.author, target.follower, False)
|
handle_author_follower_change(connection, target.author, target.follower, False)
|
||||||
)
|
)
|
||||||
|
@ -61,7 +61,7 @@ async def update_follows_for_user(connection, user_id, entity_type, entity, is_i
|
||||||
],
|
],
|
||||||
}
|
}
|
||||||
if is_insert:
|
if is_insert:
|
||||||
follows[f"{entity_type}s"].append(entity)
|
follows[f"{entity_type}s"].append(entity.dict())
|
||||||
else:
|
else:
|
||||||
# Remove the entity from follows
|
# Remove the entity from follows
|
||||||
follows[f"{entity_type}s"] = [
|
follows[f"{entity_type}s"] = [
|
||||||
|
@ -125,21 +125,11 @@ class FollowsCached:
|
||||||
redis_key = f"user:{author.user}:author"
|
redis_key = f"user:{author.user}:author"
|
||||||
author_dict = author.dict()
|
author_dict = author.dict()
|
||||||
if isinstance(author_dict, dict):
|
if isinstance(author_dict, dict):
|
||||||
filtered_author_dict = {
|
await redis.execute("set", redis_key, json.dumps(author_dict))
|
||||||
k: v for k, v in author_dict.items() if v is not None
|
|
||||||
}
|
|
||||||
await redis.execute(
|
|
||||||
"set", redis_key, json.dumps(filtered_author_dict)
|
|
||||||
)
|
|
||||||
follows = await get_author_follows(None, None, user=author.user)
|
follows = await get_author_follows(None, None, user=author.user)
|
||||||
if isinstance(follows, dict):
|
if isinstance(follows, dict):
|
||||||
filtered_follows = {
|
|
||||||
k: v for k, v in follows.items() if v is not None
|
|
||||||
}
|
|
||||||
redis_key = f"user:{author.user}:follows"
|
redis_key = f"user:{author.user}:follows"
|
||||||
await redis.execute(
|
await redis.execute("set", redis_key, json.dumps(follows))
|
||||||
"set", redis_key, json.dumps(filtered_follows)
|
|
||||||
)
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def worker():
|
async def worker():
|
||||||
|
|
Loading…
Reference in New Issue
Block a user