Add some basic lexicon cli commands
This commit is contained in:
parent
398b5705f1
commit
562d7d8a4b
|
@ -3,11 +3,11 @@ Lexicon query interface
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import re
|
import re
|
||||||
from typing import Sequence
|
from typing import Sequence, Optional
|
||||||
|
|
||||||
from sqlalchemy import select, func
|
from sqlalchemy import select, func
|
||||||
|
|
||||||
from amanuensis.db import DbContext, Lexicon
|
from amanuensis.db import DbContext, Lexicon, Membership
|
||||||
from amanuensis.errors import ArgumentError
|
from amanuensis.errors import ArgumentError
|
||||||
|
|
||||||
|
|
||||||
|
@ -17,7 +17,7 @@ RE_ALPHANUM_DASH_UNDER = re.compile(r"^[A-Za-z0-9-_]*$")
|
||||||
def create(
|
def create(
|
||||||
db: DbContext,
|
db: DbContext,
|
||||||
name: str,
|
name: str,
|
||||||
title: str,
|
title: Optional[str],
|
||||||
prompt: str,
|
prompt: str,
|
||||||
) -> Lexicon:
|
) -> Lexicon:
|
||||||
"""
|
"""
|
||||||
|
@ -55,6 +55,21 @@ def create(
|
||||||
return new_lexicon
|
return new_lexicon
|
||||||
|
|
||||||
|
|
||||||
|
def from_name(db: DbContext, name: str) -> Lexicon:
|
||||||
|
"""Get a lexicon by its name."""
|
||||||
|
return db(select(Lexicon).where(Lexicon.name == name)).scalar_one()
|
||||||
|
|
||||||
|
|
||||||
def get_all(db: DbContext) -> Sequence[Lexicon]:
|
def get_all(db: DbContext) -> Sequence[Lexicon]:
|
||||||
"""Get all lexicons."""
|
"""Get all lexicons."""
|
||||||
return db(select(Lexicon)).scalars()
|
return db(select(Lexicon)).scalars()
|
||||||
|
|
||||||
|
|
||||||
|
def get_joined(db: DbContext, user_id: int) -> Sequence[Lexicon]:
|
||||||
|
"""Get all lexicons that a player is in."""
|
||||||
|
return db(select(Lexicon).join(Lexicon.memberships).where(Membership.user_id == user_id)).scalars()
|
||||||
|
|
||||||
|
|
||||||
|
def get_public(db: DbContext) -> Sequence[Lexicon]:
|
||||||
|
"""Get all publicly visible lexicons."""
|
||||||
|
return db(select(Lexicon).where(Lexicon.public == True)).scalars()
|
||||||
|
|
|
@ -1,5 +1,13 @@
|
||||||
|
from argparse import BooleanOptionalAction
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from sqlalchemy import update
|
||||||
|
|
||||||
|
import amanuensis.backend.lexicon as lexiq
|
||||||
|
import amanuensis.backend.membership as memq
|
||||||
|
import amanuensis.backend.user as userq
|
||||||
|
from amanuensis.db import DbContext, Lexicon
|
||||||
|
|
||||||
from .helpers import add_argument
|
from .helpers import add_argument
|
||||||
|
|
||||||
|
|
||||||
|
@ -9,22 +17,51 @@ COMMAND_HELP = "Interact with lexicons."
|
||||||
LOG = logging.getLogger(__name__)
|
LOG = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@add_argument("lexicon")
|
||||||
|
@add_argument("user")
|
||||||
|
@add_argument("--editor", action="store_true")
|
||||||
|
def command_add(args) -> int:
|
||||||
|
db: DbContext = args.get_db()
|
||||||
|
lexicon = lexiq.from_name(db, args.lexicon)
|
||||||
|
user = userq.from_username(db, args.user)
|
||||||
|
assert user is not None
|
||||||
|
memq.create(db, user.id, lexicon.id, args.editor)
|
||||||
|
LOG.info(f"Added {args.user} to lexicon {args.lexicon}")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
@add_argument("name")
|
||||||
def command_create(args):
|
def command_create(args):
|
||||||
"""
|
"""
|
||||||
Create a lexicon.
|
Create a lexicon.
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError()
|
db: DbContext = args.get_db()
|
||||||
|
lexiq.create(db, args.name, None, f"Prompt for Lexicon {args.name}")
|
||||||
|
LOG.info(f"Created lexicon {args.name}")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
def command_delete(args):
|
@add_argument("name")
|
||||||
|
@add_argument("--public", action=BooleanOptionalAction)
|
||||||
|
@add_argument("--join", action=BooleanOptionalAction)
|
||||||
|
def command_edit(args):
|
||||||
"""
|
"""
|
||||||
Delete a lexicon.
|
Update a lexicon's configuration.
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError()
|
db: DbContext = args.get_db()
|
||||||
|
values = {}
|
||||||
|
|
||||||
|
if args.public == True:
|
||||||
|
values["public"] = True
|
||||||
|
elif args.public == False:
|
||||||
|
values["public"] = False
|
||||||
|
|
||||||
def command_list(args):
|
if args.join == True:
|
||||||
"""
|
values["joinable"] = True
|
||||||
List all lexicons and their statuses.
|
elif args.join == False:
|
||||||
"""
|
values["joinable"] = False
|
||||||
raise NotImplementedError()
|
|
||||||
|
result = db(update(Lexicon).where(Lexicon.name == args.name).values(**values))
|
||||||
|
LOG.info(f"Updated {result.rowcount} lexicons")
|
||||||
|
db.session.commit()
|
||||||
|
return 0 if result.rowcount == 1 else -1
|
||||||
|
|
Loading…
Reference in New Issue