aiohttp-try
This commit is contained in:
parent
36ab83d02f
commit
63eb952655
|
@ -1,11 +1,13 @@
|
|||
[0.2.16]
|
||||
- resolvers: collab inviting logics
|
||||
- orm: invite entity
|
||||
- schema: Reaction.range -> Reaction.quote
|
||||
- resolvers: queries and mutations revision and renaming
|
||||
- resolvers: delete_topic(slug) implemented
|
||||
- resolvers: added get_shout_followers
|
||||
- resolvers: load_shouts_by filters implemented
|
||||
- orm: invite entity
|
||||
- schema: Reaction.range -> Reaction.quote
|
||||
- filters: time_ago -> after
|
||||
- httpx -> aiohttp
|
||||
|
||||
[0.2.15]
|
||||
- schema: Shout.created_by removed
|
||||
|
|
9
main.py
9
main.py
|
@ -7,6 +7,7 @@ from sentry_sdk.integrations.ariadne import AriadneIntegration
|
|||
from sentry_sdk.integrations.redis import RedisIntegration
|
||||
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
|
||||
from sentry_sdk.integrations.starlette import StarletteIntegration
|
||||
from sentry_sdk.integrations.aiohttp import AioHttpIntegration
|
||||
from starlette.applications import Starlette
|
||||
from starlette.endpoints import HTTPEndpoint, Request
|
||||
from starlette.responses import JSONResponse
|
||||
|
@ -37,7 +38,13 @@ async def start_up():
|
|||
sentry_sdk.init(
|
||||
SENTRY_DSN,
|
||||
enable_tracing=True,
|
||||
integrations=[StarletteIntegration(), AriadneIntegration(), SqlalchemyIntegration(), RedisIntegration()],
|
||||
integrations=[
|
||||
StarletteIntegration(),
|
||||
AriadneIntegration(),
|
||||
SqlalchemyIntegration(),
|
||||
RedisIntegration(),
|
||||
AioHttpIntegration(),
|
||||
],
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
|
|
|
@ -9,14 +9,14 @@ readme = "README.md"
|
|||
[tool.poetry.dependencies]
|
||||
python = "^3.12"
|
||||
SQLAlchemy = "^2.0.22"
|
||||
httpx = "^0.25.0"
|
||||
psycopg2-binary = "^2.9.9"
|
||||
redis = {extras = ["hiredis"], version = "^5.0.1"}
|
||||
uvicorn = "^0.24"
|
||||
sentry-sdk = "^1.32.0"
|
||||
gql = {git = "https://github.com/graphql-python/gql.git", rev = "master"}
|
||||
starlette = {git = "https://github.com/encode/starlette.git", rev = "master"}
|
||||
ariadne = {git = "https://github.com/tonyrewin/ariadne.git", rev = "master"}
|
||||
starlette = "^0.32.0.post1"
|
||||
gql = "^3.4.1"
|
||||
ariadne = "^0.21"
|
||||
aiohttp = "^3.9.1"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
setuptools = "^69.0.2"
|
||||
|
|
|
@ -321,7 +321,7 @@ async def load_reactions_by(_, info, by, limit=50, offset=0):
|
|||
:topic - to filter by topic
|
||||
:search - to search by reactions' body
|
||||
:comment - true if body.length > 0
|
||||
:time_ago - amount of time ago
|
||||
:after - amount of time ago
|
||||
:sort - a fieldname to sort desc by default
|
||||
}
|
||||
:param limit: int amount of shouts
|
||||
|
@ -353,8 +353,8 @@ async def load_reactions_by(_, info, by, limit=50, offset=0):
|
|||
if len(by.get("search", "")) > 2:
|
||||
q = q.filter(Reaction.body.ilike(f'%{by["body"]}%'))
|
||||
|
||||
if by.get("time_ago"):
|
||||
after = int(time.time()) - int(by.get("time_ago", 0))
|
||||
if by.get("after"):
|
||||
after = int(time.time()) - int(by.get("after", 0))
|
||||
q = q.filter(Reaction.created_at > after)
|
||||
|
||||
order_way = asc if by.get("sort", "").startswith("-") else desc
|
||||
|
|
|
@ -63,9 +63,9 @@ def apply_filters(q, filters, author_id=None):
|
|||
q = q.filter(Shout.title.ilike(f'%{filters.get("title")}%'))
|
||||
if filters.get("body"):
|
||||
q = q.filter(Shout.body.ilike(f'%{filters.get("body")}%s'))
|
||||
if filters.get("time_ago"):
|
||||
before = int(time.time()) - int(filters.get("time_ago"))
|
||||
q = q.filter(Shout.created_at > before)
|
||||
if filters.get("after"):
|
||||
ts = int(filters.get("after"))
|
||||
q = q.filter(Shout.created_at > ts)
|
||||
|
||||
return q
|
||||
|
||||
|
@ -126,7 +126,7 @@ async def load_shouts_by(_, info, options):
|
|||
visibility: "public",
|
||||
author: 'discours',
|
||||
topic: 'culture',
|
||||
time_ago: 1234567 // unixtime
|
||||
after: 1234567 // unixtime
|
||||
}
|
||||
offset: 0
|
||||
limit: 50
|
||||
|
@ -166,9 +166,9 @@ async def load_shouts_by(_, info, options):
|
|||
}[filters.get("visibility")]
|
||||
if by_visibility:
|
||||
q = q.filter(Shout.visibility > by_visibility)
|
||||
by_time_ago = filters.get("time_ago")
|
||||
if by_time_ago:
|
||||
q = q.filter(Shout.created_at < by_time_ago)
|
||||
after = filters.get("after")
|
||||
if after:
|
||||
q = q.filter(Shout.created_at > after)
|
||||
|
||||
order_by = options.get("order_by", Shout.published_at)
|
||||
query_order_by = desc(order_by) if options.get("order_by_desc", True) else asc(order_by)
|
||||
|
|
|
@ -238,7 +238,7 @@ input AuthorsBy {
|
|||
name: String
|
||||
topic: String
|
||||
order: String
|
||||
time_ago: Int
|
||||
after: Int
|
||||
stat: String
|
||||
}
|
||||
|
||||
|
@ -252,7 +252,7 @@ input ShoutsFilterBy {
|
|||
authors: [String]
|
||||
layouts: [String]
|
||||
visibility: String
|
||||
time_ago: Int
|
||||
after: Int
|
||||
stat: String
|
||||
}
|
||||
|
||||
|
@ -261,7 +261,7 @@ input LoadShoutsFilters {
|
|||
author: String
|
||||
layouts: [String]
|
||||
visibility: String
|
||||
time_ago: Int
|
||||
after: Int
|
||||
}
|
||||
|
||||
input LoadShoutsOptions {
|
||||
|
@ -280,7 +280,7 @@ input ReactionBy {
|
|||
comment: Boolean
|
||||
topic: String
|
||||
created_by: Int
|
||||
time_ago: Int
|
||||
after: Int
|
||||
sort: String
|
||||
}
|
||||
|
||||
|
|
|
@ -1,5 +1,6 @@
|
|||
from functools import wraps
|
||||
from httpx import AsyncClient, HTTPError
|
||||
import aiohttp
|
||||
from aiohttp.web import HTTPUnauthorized
|
||||
from settings import AUTH_URL
|
||||
|
||||
|
||||
|
@ -19,19 +20,19 @@ async def check_auth(req):
|
|||
"variables": None,
|
||||
}
|
||||
|
||||
async with AsyncClient(timeout=30.0) as client:
|
||||
response = await client.post(AUTH_URL, headers=headers, json=gql)
|
||||
print(f"[services.auth] response: {response.status_code} {response.text}")
|
||||
if response.status_code != 200:
|
||||
return False, None
|
||||
r = response.json()
|
||||
try:
|
||||
user_id = r.get("data", {}).get(query_name, {}).get("user", {}).get("id", None)
|
||||
is_authenticated = user_id is not None
|
||||
return is_authenticated, user_id
|
||||
except Exception as e:
|
||||
print(f"{e}: {r}")
|
||||
return False, None
|
||||
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30.0)) as session:
|
||||
async with session.post(AUTH_URL, headers=headers, json=gql) as response:
|
||||
print(f"[services.auth] response: {response.status} {await response.text()}")
|
||||
if response.status != 200:
|
||||
return False, None
|
||||
r = await response.json()
|
||||
try:
|
||||
user_id = r.get("data", {}).get(query_name, {}).get("user", {}).get("id", None)
|
||||
is_authenticated = user_id is not None
|
||||
return is_authenticated, user_id
|
||||
except Exception as e:
|
||||
print(f"{e}: {r}")
|
||||
return False, None
|
||||
|
||||
|
||||
def login_required(f):
|
||||
|
@ -44,10 +45,10 @@ def login_required(f):
|
|||
if not is_authenticated:
|
||||
raise Exception("You are not logged in")
|
||||
else:
|
||||
# Добавляем author_id в контекст
|
||||
# Add user_id to the context
|
||||
context["user_id"] = user_id
|
||||
|
||||
# Если пользователь аутентифицирован, выполняем резолвер
|
||||
# If the user is authenticated, execute the resolver
|
||||
return await f(*args, **kwargs)
|
||||
|
||||
return decorated_function
|
||||
|
@ -59,7 +60,7 @@ def auth_request(f):
|
|||
req = args[0]
|
||||
is_authenticated, user_id = await check_auth(req)
|
||||
if not is_authenticated:
|
||||
raise HTTPError("please, login first")
|
||||
raise HTTPUnauthorized(text="Please, login first")
|
||||
else:
|
||||
req["user_id"] = user_id
|
||||
return await f(*args, **kwargs)
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
import asyncio
|
||||
import json
|
||||
import httpx
|
||||
import aiohttp
|
||||
from services.rediscache import redis
|
||||
from orm.shout import Shout
|
||||
|
||||
|
@ -20,13 +20,13 @@ class SearchService:
|
|||
cached = await redis.execute("GET", text)
|
||||
if not cached:
|
||||
async with SearchService.lock:
|
||||
# Use httpx to send a request to ElasticSearch
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Use aiohttp to send a request to ElasticSearch
|
||||
async with aiohttp.ClientSession() as session:
|
||||
search_url = f"https://search.discours.io/search?q={text}"
|
||||
response = await client.get(search_url)
|
||||
if response.status_code == 200:
|
||||
payload = response.json()
|
||||
await redis.execute("SET", text, payload)
|
||||
return json.loads(payload)
|
||||
async with session.get(search_url) as response:
|
||||
if response.status == 200:
|
||||
payload = await response.json()
|
||||
await redis.execute("SET", text, json.dumps(payload))
|
||||
return payload
|
||||
else:
|
||||
return json.loads(cached)
|
||||
|
|
|
@ -4,7 +4,7 @@ from datetime import timedelta, timezone, datetime
|
|||
from os import environ
|
||||
|
||||
from gql import Client, gql
|
||||
from gql.transport.httpx import HTTPXAsyncTransport
|
||||
from gql.transport.aiohttp import AIOHTTPTransport
|
||||
|
||||
from services.db import local_session
|
||||
from orm.topic import Topic
|
||||
|
@ -44,13 +44,11 @@ token = environ.get("ACKEE_TOKEN", "")
|
|||
|
||||
|
||||
def create_client(headers=None, schema=None):
|
||||
return Client(
|
||||
schema=schema,
|
||||
transport=HTTPXAsyncTransport(
|
||||
url="https://ackee.discours.io/api",
|
||||
headers=headers,
|
||||
),
|
||||
transport = AIOHTTPTransport(
|
||||
url="https://ackee.discours.io/api",
|
||||
headers=headers,
|
||||
)
|
||||
return Client(schema=schema, transport=transport)
|
||||
|
||||
|
||||
class ViewedStorage:
|
||||
|
@ -73,7 +71,7 @@ class ViewedStorage:
|
|||
async with self.lock:
|
||||
if token:
|
||||
self.client = create_client({"Authorization": "Bearer %s" % str(token)}, schema=schema_str)
|
||||
print("[services.viewed] * authorized permanentely by ackee.discours.io: %s" % token)
|
||||
print("[services.viewed] * authorized permanently by ackee.discours.io: %s" % token)
|
||||
else:
|
||||
print("[services.viewed] * please set ACKEE_TOKEN")
|
||||
self.disabled = True
|
||||
|
@ -85,19 +83,20 @@ class ViewedStorage:
|
|||
start = time.time()
|
||||
self = ViewedStorage
|
||||
try:
|
||||
self.pages = await self.client.execute_async(load_pages)
|
||||
self.pages = self.pages["domains"][0]["statistics"]["pages"]
|
||||
shouts = {}
|
||||
try:
|
||||
for page in self.pages:
|
||||
p = page["value"].split("?")[0]
|
||||
slug = p.split("discours.io/")[-1]
|
||||
shouts[slug] = page["count"]
|
||||
for slug in shouts.keys():
|
||||
await ViewedStorage.increment(slug, shouts[slug])
|
||||
except Exception:
|
||||
pass
|
||||
print("[services.viewed] ⎪ %d pages collected " % len(shouts.keys()))
|
||||
async with self.client as session:
|
||||
self.pages = await session.execute(load_pages)
|
||||
self.pages = self.pages["domains"][0]["statistics"]["pages"]
|
||||
shouts = {}
|
||||
try:
|
||||
for page in self.pages:
|
||||
p = page["value"].split("?")[0]
|
||||
slug = p.split("discours.io/")[-1]
|
||||
shouts[slug] = page["count"]
|
||||
for slug in shouts.keys():
|
||||
await ViewedStorage.increment(slug, shouts[slug])
|
||||
except Exception:
|
||||
pass
|
||||
print("[services.viewed] ⎪ %d pages collected " % len(shouts.keys()))
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
|
@ -108,7 +107,7 @@ class ViewedStorage:
|
|||
async def get_facts():
|
||||
self = ViewedStorage
|
||||
async with self.lock:
|
||||
return self.client.execute_async(load_facts)
|
||||
return await self.client.execute(load_facts)
|
||||
|
||||
@staticmethod
|
||||
async def get_shout(shout_slug):
|
||||
|
|
Loading…
Reference in New Issue
Block a user