WIP: Migration to async_pymongo

This commit is contained in:
2023-08-14 13:44:07 +02:00
parent 80ec8eb4f3
commit a1acaed6dd
13 changed files with 196 additions and 175 deletions

View File

@@ -47,12 +47,12 @@ async def album_create(
if 2 > len(title) > 40:
raise AlbumIncorrectError("title", "must be >2 and <40 characters.")
if col_albums.find_one({"name": name}) is not None:
if (await col_albums.find_one({"name": name})) is not None:
raise AlbumAlreadyExistsError(name)
makedirs(Path(f"data/users/{current_user.user}/albums/{name}"), exist_ok=True)
uploaded = col_albums.insert_one(
uploaded = await col_albums.insert_one(
{"user": current_user.user, "name": name, "title": title, "cover": None}
)
@@ -67,9 +67,10 @@ async def album_find(
current_user: User = Security(get_current_active_user, scopes=["albums.list"]),
):
output = {"results": []}
albums = list(col_albums.find({"user": current_user.user, "name": re.compile(q)}))
for album in albums:
async for album in col_albums.find(
{"user": current_user.user, "name": re.compile(q)}
):
output["results"].append(
{
"id": album["_id"].__str__(),
@@ -102,11 +103,11 @@ async def album_patch(
current_user: User = Security(get_current_active_user, scopes=["albums.write"]),
):
try:
album = col_albums.find_one({"_id": ObjectId(id)})
album = await col_albums.find_one({"_id": ObjectId(id)})
if album is None:
raise InvalidId(id)
except InvalidId:
raise AlbumNotFoundError(id)
except InvalidId as exc:
raise AlbumNotFoundError(id) from exc
if title is None:
title = album["title"]
@@ -125,7 +126,7 @@ async def album_patch(
Path(f"data/users/{current_user.user}/albums/{album['name']}"),
Path(f"data/users/{current_user.user}/albums/{name}"),
)
col_photos.update_many(
await col_photos.update_many(
{"user": current_user.user, "album": album["name"]},
{"$set": {"album": name}},
)
@@ -133,12 +134,14 @@ async def album_patch(
name = album["name"]
if cover is not None:
image = col_photos.find_one({"_id": ObjectId(cover), "album": album["name"]})
image = await col_photos.find_one(
{"_id": ObjectId(cover), "album": album["name"]}
)
cover = image["_id"].__str__() if image is not None else album["cover"]
else:
cover = album["cover"]
col_albums.update_one(
await col_albums.update_one(
{"_id": ObjectId(id)}, {"$set": {"name": name, "title": title, "cover": cover}}
)
@@ -166,11 +169,11 @@ async def album_put(
current_user: User = Security(get_current_active_user, scopes=["albums.write"]),
):
try:
album = col_albums.find_one({"_id": ObjectId(id)})
album = await col_albums.find_one({"_id": ObjectId(id)})
if album is None:
raise InvalidId(id)
except InvalidId:
raise AlbumNotFoundError(id)
except InvalidId as exc:
raise AlbumNotFoundError(id) from exc
if re.search(re.compile("^[a-z,0-9,_]*$"), name) is False:
raise AlbumIncorrectError("name", "can only contain a-z, 0-9 and _ characters.")
@@ -181,7 +184,7 @@ async def album_put(
if 2 > len(title) > 40:
raise AlbumIncorrectError("title", "must be >2 and <40 characters.")
image = col_photos.find_one({"_id": ObjectId(cover), "album": album["name"]})
image = await col_photos.find_one({"_id": ObjectId(cover), "album": album["name"]})
cover = image["_id"].__str__() if image is not None else None # type: ignore
rename(
@@ -189,10 +192,10 @@ async def album_put(
Path(f"data/users/{current_user.user}/albums/{name}"),
)
col_photos.update_many(
await col_photos.update_many(
{"user": current_user.user, "album": album["name"]}, {"$set": {"album": name}}
)
col_albums.update_one(
await col_albums.update_one(
{"_id": ObjectId(id)}, {"$set": {"name": name, "title": title, "cover": cover}}
)
@@ -213,13 +216,13 @@ async def album_delete(
current_user: User = Security(get_current_active_user, scopes=["albums.write"]),
):
try:
album = col_albums.find_one_and_delete({"_id": ObjectId(id)})
album = await col_albums.find_one_and_delete({"_id": ObjectId(id)})
if album is None:
raise InvalidId(id)
except InvalidId:
raise AlbumNotFoundError(id)
except InvalidId as exc:
raise AlbumNotFoundError(id) from exc
col_photos.delete_many({"album": album["name"]})
await col_photos.delete_many({"album": album["name"]})
rmtree(Path(f"data/users/{current_user.user}/albums/{album['name']}"))

View File

@@ -122,7 +122,7 @@ async def photo_upload(
caption: Union[str, None] = None,
current_user: User = Security(get_current_active_user, scopes=["photos.write"]),
):
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
makedirs(Path(f"data/users/{current_user.user}/albums/{album}"), exist_ok=True)
@@ -158,7 +158,7 @@ async def photo_upload(
expires_delta=timedelta(hours=configGet("media_token_valid_hours")),
)
access_token_short = uuid4().hex[:12].lower()
col_tokens.insert_one(
await col_tokens.insert_one(
{
"short": access_token_short,
"access_token": access_token,
@@ -183,7 +183,7 @@ async def photo_upload(
except (UnpackError, ValueError):
coords = {"lng": 0.0, "lat": 0.0, "alt": 0.0}
uploaded = col_photos.insert_one(
uploaded = await col_photos.insert_one(
{
"user": current_user.user,
"album": album,
@@ -231,7 +231,7 @@ if configGet("media_token_access") is True:
responses=photo_get_token_responses,
)
async def photo_get_token(token: str, id: int):
db_entry = col_tokens.find_one({"short": token})
db_entry = await col_tokens.find_one({"short": token})
if db_entry is None:
raise AccessTokenInvalidError()
@@ -246,24 +246,23 @@ if configGet("media_token_access") is True:
raise AccessTokenInvalidError()
token_scopes = payload.get("scopes", [])
token_data = TokenData(scopes=token_scopes, user=user)
except (JWTError, ValidationError) as exp:
print(exp, flush=True)
raise AccessTokenInvalidError()
except (JWTError, ValidationError) as exc:
raise AccessTokenInvalidError() from exc
user = get_user(user=token_data.user)
user_record = await get_user(user=token_data.user)
if id not in payload.get("allowed", []):
raise AccessTokenInvalidError()
try:
image = col_photos.find_one({"_id": ObjectId(id)})
image = await col_photos.find_one({"_id": ObjectId(id)})
if image is None:
raise InvalidId(id)
except InvalidId:
raise PhotoNotFoundError(id)
except InvalidId as exc:
raise PhotoNotFoundError(id) from exc
image_path = Path(
f"data/users/{user.user}/albums/{image['album']}/{image['filename']}"
f"data/users/{user_record.user}/albums/{image['album']}/{image['filename']}"
)
mime = Magic(mime=True).from_file(image_path)
@@ -301,11 +300,11 @@ async def photo_get(
current_user: User = Security(get_current_active_user, scopes=["photos.read"]),
):
try:
image = col_photos.find_one({"_id": ObjectId(id)})
image = await col_photos.find_one({"_id": ObjectId(id)})
if image is None:
raise InvalidId(id)
except InvalidId:
raise PhotoNotFoundError(id)
except InvalidId as exc:
raise PhotoNotFoundError(id) from exc
image_path = Path(
f"data/users/{current_user.user}/albums/{image['album']}/{image['filename']}"
@@ -334,13 +333,13 @@ async def photo_move(
current_user: User = Security(get_current_active_user, scopes=["photos.write"]),
):
try:
image = col_photos.find_one({"_id": ObjectId(id)})
image = await col_photos.find_one({"_id": ObjectId(id)})
if image is None:
raise InvalidId(id)
except InvalidId:
raise PhotoNotFoundError(id)
except InvalidId as exc:
raise PhotoNotFoundError(id) from exc
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
if Path(
@@ -354,7 +353,7 @@ async def photo_move(
else:
filename = image["filename"]
col_photos.find_one_and_update(
await col_photos.find_one_and_update(
{"_id": ObjectId(id)},
{
"$set": {
@@ -396,13 +395,13 @@ async def photo_patch(
current_user: User = Security(get_current_active_user, scopes=["photos.write"]),
):
try:
image = col_photos.find_one({"_id": ObjectId(id)})
image = await col_photos.find_one({"_id": ObjectId(id)})
if image is None:
raise InvalidId(id)
except InvalidId:
raise PhotoNotFoundError(id)
except InvalidId as exc:
raise PhotoNotFoundError(id) from exc
col_photos.find_one_and_update(
await col_photos.find_one_and_update(
{"_id": ObjectId(id)},
{"$set": {"caption": caption, "dates.modified": datetime.now(tz=timezone.utc)}},
)
@@ -430,16 +429,16 @@ async def photo_delete(
current_user: User = Security(get_current_active_user, scopes=["photos.write"]),
):
try:
image = col_photos.find_one_and_delete({"_id": ObjectId(id)})
image = await col_photos.find_one_and_delete({"_id": ObjectId(id)})
if image is None:
raise InvalidId(id)
except InvalidId:
raise PhotoNotFoundError(id)
except InvalidId as exc:
raise PhotoNotFoundError(id) from exc
album = col_albums.find_one({"name": image["album"]})
album = await col_albums.find_one({"name": image["album"]})
if album is not None and album["cover"] == image["_id"].__str__():
col_albums.update_one({"name": image["album"]}, {"$set": {"cover": None}})
await col_albums.update_one({"name": image["album"]}, {"$set": {"cover": None}})
remove(
Path(
@@ -469,7 +468,7 @@ async def photo_random(
limit: int = 100,
current_user: User = Security(get_current_active_user, scopes=["photos.list"]),
):
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
if limit <= 0:
@@ -490,20 +489,16 @@ async def photo_random(
}
)
documents_count = col_photos.count_documents(db_query)
documents_count = await col_photos.count_documents(db_query)
skip = randint(0, documents_count - 1) if documents_count > 1 else 0
images = list(
col_photos.aggregate(
[
{"$match": db_query},
{"$skip": skip},
{"$limit": limit},
]
)
)
for image in images:
async for image in col_photos.aggregate(
[
{"$match": db_query},
{"$skip": skip},
{"$limit": limit},
]
):
output["results"].append(
{
"id": image["_id"].__str__(),
@@ -543,7 +538,7 @@ async def photo_find(
current_user: User = Security(get_current_active_user, scopes=["photos.list"]),
):
if token is not None:
found_record = col_tokens.find_one({"token": token})
found_record = await col_tokens.find_one({"token": token})
if found_record is None:
raise SearchTokenInvalidError()
@@ -560,7 +555,7 @@ async def photo_find(
current_user=current_user,
)
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
if page <= 0 or page_size <= 0:
@@ -612,16 +607,22 @@ async def photo_find(
"filename": re.compile(q),
}
else:
db_query = {"user": current_user.user, "album": album, "filename": re.compile(q), "caption": re.compile(caption)} # type: ignore
db_query_count = {"user": current_user.user, "album": album, "filename": re.compile(q), "caption": re.compile(caption)} # type: ignore
db_query = {
"user": current_user.user,
"album": album,
"filename": re.compile(q),
"caption": re.compile(caption),
}
db_query_count = {
"user": current_user.user,
"album": album,
"filename": re.compile(q),
"caption": re.compile(caption),
}
images = list(
col_photos.find(db_query, limit=page_size, skip=skip).sort(
"dates.uploaded", DESCENDING
)
)
for image in images:
async for image in col_photos.find(db_query, limit=page_size, skip=skip).sort(
"dates.uploaded", DESCENDING
):
output["results"].append(
{
"id": image["_id"].__str__(),
@@ -630,9 +631,9 @@ async def photo_find(
}
)
if col_photos.count_documents(db_query_count) > page * page_size:
if (await col_photos.count_documents(db_query_count)) > page * page_size:
token = str(token_urlsafe(32))
col_tokens.insert_one(
await col_tokens.insert_one(
{
"token": token,
"query": q,

View File

@@ -17,7 +17,7 @@ token_post_responses = {401: UserCredentialsInvalid().openapi}
@app.post("/token", response_model=Token, responses=token_post_responses)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form_data.username, form_data.password)
user = await authenticate_user(form_data.username, form_data.password)
if not user:
raise UserCredentialsInvalid()
access_token_expires = timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS)

View File

@@ -41,14 +41,14 @@ async def send_confirmation(user: str, email: str):
+ f"/users/{user}/confirm?code={confirmation_code}"
),
)
col_emails.insert_one(
await col_emails.insert_one(
{"user": user, "email": email, "used": False, "code": confirmation_code}
)
logger.info(
"Sent confirmation email to '%s' with code %s", email, confirmation_code
)
except Exception as exp:
logger.error("Could not send confirmation email to '%s' due to: %s", email, exp)
except Exception as exc:
logger.error("Could not send confirmation email to '%s' due to: %s", email, exc)
@app.get("/users/me/", response_model=User)
@@ -80,15 +80,15 @@ if configGet("registration_requires_confirmation") is True:
responses=user_confirm_responses,
)
async def user_confirm(user: str, code: str):
confirm_record = col_emails.find_one(
confirm_record = await col_emails.find_one(
{"user": user, "code": code, "used": False}
)
if confirm_record is None:
raise UserEmailCodeInvalid()
col_emails.find_one_and_update(
await col_emails.find_one_and_update(
{"_id": confirm_record["_id"]}, {"$set": {"used": True}}
)
col_users.find_one_and_update(
await col_users.find_one_and_update(
{"user": confirm_record["user"]}, {"$set": {"disabled": False}}
)
return UJSONResponse({"detail": configGet("email_confirmed", "messages")})
@@ -103,9 +103,9 @@ if configGet("registration_enabled") is True:
async def user_create(
user: str = Form(), email: str = Form(), password: str = Form()
):
if col_users.find_one({"user": user}) is not None:
if (await col_users.find_one({"user": user})) is not None:
raise UserAlreadyExists()
col_users.insert_one(
await col_users.insert_one(
{
"user": user,
"email": email,
@@ -132,14 +132,14 @@ user_delete_responses = {401: UserCredentialsInvalid().openapi}
async def user_delete(
password: str = Form(), current_user: User = Depends(get_current_active_user)
):
user = get_user(current_user.user)
user = await get_user(current_user.user)
if not user:
return False
if not verify_password(password, user.hash):
raise UserCredentialsInvalid()
col_users.delete_many({"user": current_user.user})
col_emails.delete_many({"user": current_user.user})
col_photos.delete_many({"user": current_user.user})
col_videos.delete_many({"user": current_user.user})
col_albums.delete_many({"user": current_user.user})
await col_users.delete_many({"user": current_user.user})
await col_emails.delete_many({"user": current_user.user})
await col_photos.delete_many({"user": current_user.user})
await col_videos.delete_many({"user": current_user.user})
await col_albums.delete_many({"user": current_user.user})
return Response(status_code=HTTP_204_NO_CONTENT)

View File

@@ -50,7 +50,7 @@ async def video_upload(
caption: Union[str, None] = None,
current_user: User = Security(get_current_active_user, scopes=["videos.write"]),
):
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
makedirs(Path(f"data/users/{current_user.user}/albums/{album}"), exist_ok=True)
@@ -73,7 +73,7 @@ async def video_upload(
# Coords extraction should be here
uploaded = col_videos.insert_one(
uploaded = await col_videos.insert_one(
{
"user": current_user.user,
"album": album,
@@ -123,11 +123,11 @@ async def video_get(
current_user: User = Security(get_current_active_user, scopes=["videos.read"]),
):
try:
video = col_videos.find_one({"_id": ObjectId(id)})
video = await col_videos.find_one({"_id": ObjectId(id)})
if video is None:
raise InvalidId(id)
except InvalidId:
raise VideoNotFoundError(id)
except InvalidId as exc:
raise VideoNotFoundError(id) from exc
video_path = Path(
f"data/users/{current_user.user}/albums/{video['album']}/{video['filename']}"
@@ -156,13 +156,13 @@ async def video_move(
current_user: User = Security(get_current_active_user, scopes=["videos.write"]),
):
try:
video = col_videos.find_one({"_id": ObjectId(id)})
video = await col_videos.find_one({"_id": ObjectId(id)})
if video is None:
raise InvalidId(id)
except InvalidId:
raise VideoNotFoundError(id)
except InvalidId as exc:
raise VideoNotFoundError(id) from exc
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
if Path(
@@ -176,7 +176,7 @@ async def video_move(
else:
filename = video["filename"]
col_videos.find_one_and_update(
await col_videos.find_one_and_update(
{"_id": ObjectId(id)},
{
"$set": {
@@ -218,13 +218,13 @@ async def video_patch(
current_user: User = Security(get_current_active_user, scopes=["videos.write"]),
):
try:
video = col_videos.find_one({"_id": ObjectId(id)})
video = await col_videos.find_one({"_id": ObjectId(id)})
if video is None:
raise InvalidId(id)
except InvalidId:
raise VideoNotFoundError(id)
except InvalidId as exc:
raise VideoNotFoundError(id) from exc
col_videos.find_one_and_update(
await col_videos.find_one_and_update(
{"_id": ObjectId(id)},
{"$set": {"caption": caption, "dates.modified": datetime.now(tz=timezone.utc)}},
)
@@ -252,13 +252,13 @@ async def video_delete(
current_user: User = Security(get_current_active_user, scopes=["videos.write"]),
):
try:
video = col_videos.find_one_and_delete({"_id": ObjectId(id)})
video = await col_videos.find_one_and_delete({"_id": ObjectId(id)})
if video is None:
raise InvalidId(id)
except InvalidId:
raise VideoNotFoundError(id)
except InvalidId as exc:
raise VideoNotFoundError(id) from exc
album = col_albums.find_one({"name": video["album"]})
album = await col_albums.find_one({"name": video["album"]})
remove(
Path(
@@ -288,7 +288,7 @@ async def video_random(
limit: int = 100,
current_user: User = Security(get_current_active_user, scopes=["videos.list"]),
):
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
if limit <= 0:
@@ -309,20 +309,16 @@ async def video_random(
}
)
documents_count = col_videos.count_documents(db_query)
documents_count = await col_videos.count_documents(db_query)
skip = randint(0, documents_count - 1) if documents_count > 1 else 0
videos = list(
col_videos.aggregate(
[
{"$match": db_query},
{"$skip": skip},
{"$limit": limit},
]
)
)
for video in videos:
async for video in col_videos.aggregate(
[
{"$match": db_query},
{"$skip": skip},
{"$limit": limit},
]
):
output["results"].append(
{
"id": video["_id"].__str__(),
@@ -359,7 +355,7 @@ async def video_find(
current_user: User = Security(get_current_active_user, scopes=["videos.list"]),
):
if token is not None:
found_record = col_tokens.find_one({"token": token})
found_record = await col_tokens.find_one({"token": token})
if found_record is None:
raise SearchTokenInvalidError()
@@ -373,7 +369,7 @@ async def video_find(
current_user=current_user,
)
if col_albums.find_one({"user": current_user.user, "name": album}) is None:
if (await col_albums.find_one({"user": current_user.user, "name": album})) is None:
raise AlbumNameNotFoundError(album)
if page <= 0 or page_size <= 0:
@@ -410,16 +406,28 @@ async def video_find(
"caption": re.compile(q),
}
else:
db_query = list(col_videos.find({"user": current_user.user, "album": album, "filename": re.compile(q), "caption": re.compile(caption)}, limit=page_size, skip=skip).sort("dates.uploaded", DESCENDING)) # type: ignore
db_query_count = {"user": current_user.user, "album": album, "filename": re.compile(q), "caption": re.compile(caption)} # type: ignore
videos = list(
col_videos.find(db_query, limit=page_size, skip=skip).sort(
"dates.uploaded", DESCENDING
db_query = list(
col_videos.find(
{
"user": current_user.user,
"album": album,
"filename": re.compile(q),
"caption": re.compile(caption),
},
limit=page_size,
skip=skip,
).sort("dates.uploaded", DESCENDING)
)
)
db_query_count = {
"user": current_user.user,
"album": album,
"filename": re.compile(q),
"caption": re.compile(caption),
}
for video in videos:
async for video in col_videos.find(db_query, limit=page_size, skip=skip).sort(
"dates.uploaded", DESCENDING
):
output["results"].append(
{
"id": video["_id"].__str__(),
@@ -428,9 +436,9 @@ async def video_find(
}
)
if col_videos.count_documents(db_query_count) > page * page_size:
if (await col_videos.count_documents(db_query_count)) > page * page_size:
token = str(token_urlsafe(32))
col_tokens.insert_one(
await col_tokens.insert_one(
{
"token": token,
"query": q,