# Return the object
return object
+ def fetch(self, stmt, batch_size=128):
+ """
+ Fetches objects of the given type
+ """
+ result = self.execute(stmt)
+
+ # Process the result in batches
+ if batch_size:
+ result = result.yield_per(batch_size)
+
+ # Apply unique filtering
+ result = result.unique()
+
+ # Return all objects
+ for object in result.scalars():
+ yield object
+
def fetch_one(self, stmt):
result = self.execute(stmt)
# Return exactly one object or none, but fail otherwise
return result.scalar_one_or_none()
+
+ def fetch_as_list(self, *args, **kwargs):
+ """
+ Fetches objects and returns them as a list instead of an iterator
+ """
+ objects = self.fetch(*args, **kwargs)
+
+ # Return as list
+ return [o for o in objects]
+
+ def fetch_as_set(self, *args, **kwargs):
+ """
+ Fetches objects and returns them as a set instead of an iterator
+ """
+ objects = self.fetch(*args, **kwargs)
+
+ # Return as set
+ return set([o for o in objects])
def __init__(self, backend):
self.backend = backend
+ def __iter__(self):
+ """
+ Returns an iterator over all lists
+ """
+ stmt = (
+ sqlmodel
+ .select(
+ List,
+ )
+ .where(
+ List.deleted_at == None,
+ )
+ .order_by(
+ List.name,
+ List.slug,
+ )
+ )
+
+ return self.backend.db.fetch(stmt)
+
def get_by_slug(self, slug):
stmt = (
sqlmodel
# Show Version
parser.add_argument("--version", action="version", version="%(prog)s @VERSION@")
+ # lists
+ lists = subparsers.add_parser("lists", help=_("Show all lists"))
+ lists.set_defaults(func=self.__lists)
+
# list-create
list_create = subparsers.add_parser("list-create", help=_("Create a new list"))
list_create.add_argument("--name", required=True,
# Otherwise just exit
sys.exit(0)
+ def __lists(self, backend, args):
+ table = rich.table.Table(title=_("Lists"))
+ table.add_column(_("Name"))
+ table.add_column(_("Slug"))
+ table.add_column(_("License"))
+ table.add_column(_("Created At"))
+ table.add_column(_("Created By"))
+
+ # Show all lists
+ for list in backend.lists:
+ table.add_row(
+ list.name,
+ list.slug,
+ list.license,
+ list.created_at.isoformat(),
+ list.created_by,
+ )
+
+ # Print the table
+ self.console.print(table)
+
def __list_create(self, backend, args):
"""
Creates a new list