Now with queries using mongoengine.

Author: mikeckennedy

app/ch16_mongodb/final/pypi_org/services/package_service.py

@@ -1,42 +1,23 @@
 from typing import List, Optional
-import sqlalchemy.orm
-from sqlalchemy.orm import Session
 
-import pypi_org.data.db_session as db_session
-from pypi_org.data.package import Package
-from pypi_org.data.releases import Release
+from pypi_org.nosql.packages import Package
+from pypi_org.nosql.releases import Release
 
 
 def get_latest_releases(limit=10) -> List[Release]:
-    session = db_session.create_session()
-    try:
-
-        releases = session.query(Release). \
-            options(sqlalchemy.orm.joinedload(Release.package)). \
-            order_by(Release.created_date.desc()). \
-            limit(limit). \
-            all()
-
-    finally:
-        session.close()
-
+    releases = Release.objects(). \
+        order_by("-created_date"). \
+        limit(limit). \
+        all()
     return releases
 
 
 def get_package_count() -> int:
-    session = db_session.create_session()
-    try:
-        return session.query(Package).count()
-    finally:
-        session.close()
+    return Package.objects().count()
 
 
 def get_release_count() -> int:
-    session = db_session.create_session()
-    try:
-        return session.query(Release).count()
-    finally:
-        session.close()
+    return Release.objects().count()
 
 
 def get_package_by_id(package_id: str) -> Optional[Package]:
@@ -45,23 +26,12 @@ def get_package_by_id(package_id: str) -> Optional[Package]:
 
     package_id = package_id.strip().lower()
 
-    session = db_session.create_session()
-    try:
-
-        package = session.query(Package) \
-            .options(sqlalchemy.orm.joinedload(Package.releases)) \
-            .filter(Package.id == package_id) \
-            .first()
-
-    finally:
-        session.close()
+    package = Package.objects() \
+        .filter(id=package_id) \
+        .first()
 
     return package
 
 
 def all_packages(limit: int) -> List[Package]:
-    session: Session = db_session.create_session()
-    try:
-        return list(session.query(Package).limit(limit))
-    finally:
-        session.close()
+    return list(Package.objects().limit(limit))

app/ch16_mongodb/final/pypi_org/services/user_service.py

@@ -1,24 +1,15 @@
 from typing import Optional
 
 from passlib.handlers.sha2_crypt import sha512_crypt as crypto
-import pypi_org.data.db_session as db_session
-from pypi_org.data.users import User
+from pypi_org.nosql.users import User
 
 
 def get_user_count() -> int:
-    session = db_session.create_session()
-    try:
-        return session.query(User).count()
-    finally:
-        session.close()
+    return User.objects().count()
 
 
 def find_user_by_email(email: str) -> Optional[User]:
-    session = db_session.create_session()
-    try:
-        return session.query(User).filter(User.email == email).first()
-    finally:
-        session.close()
+    return User.objects().filter(email=email).first()
 
 
 def create_user(name: str, email: str, password: str) -> Optional[User]:
@@ -30,12 +21,7 @@ def create_user(name: str, email: str, password: str) -> Optional[User]:
     user.name = name
     user.hashed_password = hash_text(password)
 
-    session = db_session.create_session()
-    try:
-        session.add(user)
-        session.commit()
-    finally:
-        session.close()
+    user.save()
 
     return user
 
@@ -50,24 +36,16 @@ def verify_hash(hashed_text: str, plain_text: str) -> bool:
 
 
 def login_user(email: str, password: str) -> Optional[User]:
-    session = db_session.create_session()
-    try:
-        user = session.query(User).filter(User.email == email).first()
-        if not user:
-            return None
+    user = find_user_by_email(email)
+    if not user:
+        return None
 
-        if not verify_hash(user.hashed_password, password):
-            return None
+    if not verify_hash(user.hashed_password, password):
+        return None
 
-        return user
-    finally:
-        session.close()
+    return user
 
 
 def find_user_by_id(user_id: int) -> Optional[User]:
-    session = db_session.create_session()
-    try:
-        user = session.query(User).filter(User.id == user_id).first()
-        return user
-    finally:
-        session.close()
+    user = User.objects().filter(id=user_id).first()
+    return user