change(fastapi): simplify model imports across code-base

Closes: #133

Signed-off-by: Kevin Morris <kevr@0cost.org>
This commit is contained in:
Kevin Morris 2021-10-16 19:25:25 -07:00
parent bfdc85d7d6
commit 28c4e9697b
No known key found for this signature in database
GPG key ID: F7E46DED420788F3
12 changed files with 341 additions and 343 deletions

View file

@ -7,43 +7,35 @@ import orjson
from fastapi import HTTPException
from sqlalchemy import and_, orm
from aurweb import db
from aurweb.models.official_provider import OFFICIAL_BASE, OfficialProvider
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.package_comment import PackageComment
from aurweb.models.package_dependency import PackageDependency
from aurweb.models.package_notification import PackageNotification
from aurweb.models.package_relation import PackageRelation
from aurweb.models.package_vote import PackageVote
from aurweb.models.relation_type import PROVIDES_ID, RelationType
from aurweb.models.user import User
from aurweb import db, models
from aurweb.models.official_provider import OFFICIAL_BASE
from aurweb.models.relation_type import PROVIDES_ID
from aurweb.redis import redis_connection
from aurweb.templates import register_filter
def dep_depends_extra(dep: PackageDependency) -> str:
def dep_depends_extra(dep: models.PackageDependency) -> str:
""" A function used to produce extra text for dependency display. """
return str()
def dep_makedepends_extra(dep: PackageDependency) -> str:
def dep_makedepends_extra(dep: models.PackageDependency) -> str:
""" A function used to produce extra text for dependency display. """
return "(make)"
def dep_checkdepends_extra(dep: PackageDependency) -> str:
def dep_checkdepends_extra(dep: models.PackageDependency) -> str:
""" A function used to produce extra text for dependency display. """
return "(check)"
def dep_optdepends_extra(dep: PackageDependency) -> str:
def dep_optdepends_extra(dep: models.PackageDependency) -> str:
""" A function used to produce extra text for dependency display. """
return "(optional)"
@register_filter("dep_extra")
def dep_extra(dep: PackageDependency) -> str:
def dep_extra(dep: models.PackageDependency) -> str:
""" Some dependency types have extra text added to their
display. This function provides that output. However, it
**assumes** that the dep passed is bound to a valid one
@ -53,7 +45,7 @@ def dep_extra(dep: PackageDependency) -> str:
@register_filter("dep_extra_desc")
def dep_extra_desc(dep: PackageDependency) -> str:
def dep_extra_desc(dep: models.PackageDependency) -> str:
extra = dep_extra(dep)
if not dep.DepDesc:
return extra
@ -63,30 +55,30 @@ def dep_extra_desc(dep: PackageDependency) -> str:
@register_filter("pkgname_link")
def pkgname_link(pkgname: str) -> str:
base = "/".join([OFFICIAL_BASE, "packages"])
official = db.query(OfficialProvider).filter(
OfficialProvider.Name == pkgname)
official = db.query(models.OfficialProvider).filter(
models.OfficialProvider.Name == pkgname)
if official.scalar():
return f"{base}/?q={pkgname}"
return f"/packages/{pkgname}"
@register_filter("package_link")
def package_link(package: Package) -> str:
def package_link(package: models.Package) -> str:
base = "/".join([OFFICIAL_BASE, "packages"])
official = db.query(OfficialProvider).filter(
OfficialProvider.Name == package.Name)
official = db.query(models.OfficialProvider).filter(
models.OfficialProvider.Name == package.Name)
if official.scalar():
return f"{base}/?q={package.Name}"
return f"/packages/{package.Name}"
@register_filter("provides_list")
def provides_list(package: Package, depname: str) -> list:
providers = db.query(Package).join(
PackageRelation).join(RelationType).filter(
def provides_list(package: models.Package, depname: str) -> list:
providers = db.query(models.Package).join(
models.PackageRelation).join(models.RelationType).filter(
and_(
PackageRelation.RelName == depname,
RelationType.ID == PROVIDES_ID
models.PackageRelation.RelName == depname,
models.RelationType.ID == PROVIDES_ID
)
)
@ -102,7 +94,9 @@ def provides_list(package: Package, depname: str) -> list:
return string
def get_pkg_or_base(name: str, cls: Union[Package, PackageBase] = PackageBase):
def get_pkg_or_base(
name: str,
cls: Union[models.Package, models.PackageBase] = models.PackageBase):
""" Get a PackageBase instance by its name or raise a 404 if
it can't be found in the database.
@ -110,20 +104,21 @@ def get_pkg_or_base(name: str, cls: Union[Package, PackageBase] = PackageBase):
:raises HTTPException: With status code 404 if record doesn't exist
:return: {Package,PackageBase} instance
"""
provider = db.query(OfficialProvider).filter(
OfficialProvider.Name == name).first()
provider = db.query(models.OfficialProvider).filter(
models.OfficialProvider.Name == name).first()
if provider:
raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
instance = db.query(cls).filter(cls.Name == name).first()
if cls == PackageBase and not instance:
if cls == models.PackageBase and not instance:
raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
return instance
def get_pkgbase_comment(pkgbase: PackageBase, id: int) -> PackageComment:
comment = pkgbase.comments.filter(PackageComment.ID == id).first()
def get_pkgbase_comment(
pkgbase: models.PackageBase, id: int) -> models.PackageComment:
comment = pkgbase.comments.filter(models.PackageComment.ID == id).first()
if not comment:
raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
return comment
@ -131,10 +126,11 @@ def get_pkgbase_comment(pkgbase: PackageBase, id: int) -> PackageComment:
@register_filter("out_of_date")
def out_of_date(packages: orm.Query) -> orm.Query:
return packages.filter(PackageBase.OutOfDateTS.isnot(None))
return packages.filter(models.PackageBase.OutOfDateTS.isnot(None))
def updated_packages(limit: int = 0, cache_ttl: int = 600) -> List[Package]:
def updated_packages(limit: int = 0,
cache_ttl: int = 600) -> List[models.Package]:
""" Return a list of valid Package objects ordered by their
ModifiedTS column in descending order from cache, after setting
the cache when no key yet exists.
@ -149,10 +145,10 @@ def updated_packages(limit: int = 0, cache_ttl: int = 600) -> List[Package]:
# If we already have a cache, deserialize it and return.
return orjson.loads(packages)
query = db.query(Package).join(PackageBase).filter(
PackageBase.PackagerUID.isnot(None)
query = db.query(models.Package).join(models.PackageBase).filter(
models.PackageBase.PackagerUID.isnot(None)
).order_by(
PackageBase.ModifiedTS.desc()
models.PackageBase.ModifiedTS.desc()
)
if limit:
@ -178,7 +174,8 @@ def updated_packages(limit: int = 0, cache_ttl: int = 600) -> List[Package]:
return packages
def query_voted(query: List[Package], user: User) -> Dict[int, bool]:
def query_voted(query: List[models.Package],
user: models.User) -> Dict[int, bool]:
""" Produce a dictionary of package base ID keys to boolean values,
which indicate whether or not the package base has a vote record
related to user.
@ -189,18 +186,19 @@ def query_voted(query: List[Package], user: User) -> Dict[int, bool]:
"""
output = defaultdict(bool)
query_set = {pkg.PackageBase.ID for pkg in query}
voted = db.query(PackageVote).join(
PackageBase,
PackageBase.ID.in_(query_set)
voted = db.query(models.PackageVote).join(
models.PackageBase,
models.PackageBase.ID.in_(query_set)
).filter(
PackageVote.UsersID == user.ID
models.PackageVote.UsersID == user.ID
)
for vote in voted:
output[vote.PackageBase.ID] = True
return output
def query_notified(query: List[Package], user: User) -> Dict[int, bool]:
def query_notified(query: List[models.Package],
user: models.User) -> Dict[int, bool]:
""" Produce a dictionary of package base ID keys to boolean values,
which indicate whether or not the package base has a notification
record related to user.
@ -211,11 +209,11 @@ def query_notified(query: List[Package], user: User) -> Dict[int, bool]:
"""
output = defaultdict(bool)
query_set = {pkg.PackageBase.ID for pkg in query}
notified = db.query(PackageNotification).join(
PackageBase,
PackageBase.ID.in_(query_set)
notified = db.query(models.PackageNotification).join(
models.PackageBase,
models.PackageBase.ID.in_(query_set)
).filter(
PackageNotification.UserID == user.ID
models.PackageNotification.UserID == user.ID
)
for notify in notified:
output[notify.PackageBase.ID] = True