fix usermanager mypy

This commit is contained in:
dni ⚡
2022-12-20 13:14:49 +01:00
parent c5fdd35078
commit c90f7878d7
5 changed files with 50 additions and 55 deletions

View File

@ -10,12 +10,10 @@ from lnbits.core.crud import (
from lnbits.core.models import Payment from lnbits.core.models import Payment
from . import db from . import db
from .models import CreateUserData, Users, Wallets from .models import CreateUserData, User, Wallet
### Users
async def create_usermanager_user(data: CreateUserData) -> Users: async def create_usermanager_user(data: CreateUserData) -> User:
account = await create_account() account = await create_account()
user = await get_user(account.id) user = await get_user(account.id)
assert user, "Newly created user couldn't be retrieved" assert user, "Newly created user couldn't be retrieved"
@ -50,17 +48,17 @@ async def create_usermanager_user(data: CreateUserData) -> Users:
return user_created return user_created
async def get_usermanager_user(user_id: str) -> Optional[Users]: async def get_usermanager_user(user_id: str) -> Optional[User]:
row = await db.fetchone("SELECT * FROM usermanager.users WHERE id = ?", (user_id,)) row = await db.fetchone("SELECT * FROM usermanager.users WHERE id = ?", (user_id,))
return Users(**row) if row else None return User(**row) if row else None
async def get_usermanager_users(user_id: str) -> List[Users]: async def get_usermanager_users(user_id: str) -> list[User]:
rows = await db.fetchall( rows = await db.fetchall(
"SELECT * FROM usermanager.users WHERE admin = ?", (user_id,) "SELECT * FROM usermanager.users WHERE admin = ?", (user_id,)
) )
return [Users(**row) for row in rows] return [User(**row) for row in rows]
async def delete_usermanager_user(user_id: str, delete_core: bool = True) -> None: async def delete_usermanager_user(user_id: str, delete_core: bool = True) -> None:
@ -73,12 +71,9 @@ async def delete_usermanager_user(user_id: str, delete_core: bool = True) -> Non
await db.execute("""DELETE FROM usermanager.wallets WHERE "user" = ?""", (user_id,)) await db.execute("""DELETE FROM usermanager.wallets WHERE "user" = ?""", (user_id,))
### Wallets
async def create_usermanager_wallet( async def create_usermanager_wallet(
user_id: str, wallet_name: str, admin_id: str user_id: str, wallet_name: str, admin_id: str
) -> Wallets: ) -> Wallet:
wallet = await create_wallet(user_id=user_id, wallet_name=wallet_name) wallet = await create_wallet(user_id=user_id, wallet_name=wallet_name)
await db.execute( await db.execute(
""" """
@ -92,28 +87,28 @@ async def create_usermanager_wallet(
return wallet_created return wallet_created
async def get_usermanager_wallet(wallet_id: str) -> Optional[Wallets]: async def get_usermanager_wallet(wallet_id: str) -> Optional[Wallet]:
row = await db.fetchone( row = await db.fetchone(
"SELECT * FROM usermanager.wallets WHERE id = ?", (wallet_id,) "SELECT * FROM usermanager.wallets WHERE id = ?", (wallet_id,)
) )
return Wallets(**row) if row else None return Wallet(**row) if row else None
async def get_usermanager_wallets(admin_id: str) -> Optional[Wallets]: async def get_usermanager_wallets(admin_id: str) -> list[Wallet]:
rows = await db.fetchall( rows = await db.fetchall(
"SELECT * FROM usermanager.wallets WHERE admin = ?", (admin_id,) "SELECT * FROM usermanager.wallets WHERE admin = ?", (admin_id,)
) )
return [Wallets(**row) for row in rows] return [Wallet(**row) for row in rows]
async def get_usermanager_users_wallets(user_id: str) -> Optional[Wallets]: async def get_usermanager_users_wallets(user_id: str) -> list[Wallet]:
rows = await db.fetchall( rows = await db.fetchall(
"""SELECT * FROM usermanager.wallets WHERE "user" = ?""", (user_id,) """SELECT * FROM usermanager.wallets WHERE "user" = ?""", (user_id,)
) )
return [Wallets(**row) for row in rows] return [Wallet(**row) for row in rows]
async def get_usermanager_wallet_transactions(wallet_id: str) -> Optional[Payment]: async def get_usermanager_wallet_transactions(wallet_id: str) -> list[Payment]:
return await get_payments( return await get_payments(
wallet_id=wallet_id, complete=True, pending=False, outgoing=True, incoming=True wallet_id=wallet_id, complete=True, pending=False, outgoing=True, incoming=True
) )

View File

@ -19,7 +19,7 @@ class CreateUserWallet(BaseModel):
admin_id: str = Query(...) admin_id: str = Query(...)
class Users(BaseModel): class User(BaseModel):
id: str id: str
name: str name: str
admin: str admin: str
@ -27,7 +27,7 @@ class Users(BaseModel):
password: Optional[str] = None password: Optional[str] = None
class Wallets(BaseModel): class Wallet(BaseModel):
id: str id: str
admin: str admin: str
name: str name: str
@ -36,5 +36,5 @@ class Wallets(BaseModel):
inkey: str inkey: str
@classmethod @classmethod
def from_row(cls, row: Row) -> "Wallets": def from_row(cls, row: Row) -> "Wallet":
return cls(**dict(row)) return cls(**dict(row))

View File

@ -9,7 +9,10 @@ from . import usermanager_ext, usermanager_renderer
@usermanager_ext.get("/", response_class=HTMLResponse) @usermanager_ext.get("/", response_class=HTMLResponse)
async def index(request: Request, user: User = Depends(check_user_exists)): async def index(
request: Request,
user: User = Depends(check_user_exists) #type: ignore
):
return usermanager_renderer().TemplateResponse( return usermanager_renderer().TemplateResponse(
"usermanager/index.html", {"request": request, "user": user.dict()} "usermanager/index.html", {"request": request, "user": user.dict()}
) )

View File

@ -21,40 +21,44 @@ from .crud import (
get_usermanager_wallet_transactions, get_usermanager_wallet_transactions,
get_usermanager_wallets, get_usermanager_wallets,
) )
from .models import CreateUserData, CreateUserWallet from .models import CreateUserData, CreateUserWallet, Wallet
# Users
@usermanager_ext.get("/api/v1/users", status_code=HTTPStatus.OK) @usermanager_ext.get("/api/v1/users", status_code=HTTPStatus.OK)
async def api_usermanager_users(wallet: WalletTypeInfo = Depends(require_admin_key)): async def api_usermanager_users(
wallet: WalletTypeInfo = Depends(require_admin_key) #type: ignore
):
user_id = wallet.wallet.user user_id = wallet.wallet.user
return [user.dict() for user in await get_usermanager_users(user_id)] return [user.dict() for user in await get_usermanager_users(user_id)]
@usermanager_ext.get("/api/v1/users/{user_id}", status_code=HTTPStatus.OK) @usermanager_ext.get("/api/v1/users/{user_id}", status_code=HTTPStatus.OK)
async def api_usermanager_user(user_id, wallet: WalletTypeInfo = Depends(get_key_type)): async def api_usermanager_user(
user_id,
wallet: WalletTypeInfo = Depends(get_key_type) #type: ignore
):
user = await get_usermanager_user(user_id) user = await get_usermanager_user(user_id)
if not user:
return None
return user.dict() return user.dict()
@usermanager_ext.post("/api/v1/users", status_code=HTTPStatus.CREATED) @usermanager_ext.post("/api/v1/users", status_code=HTTPStatus.CREATED, dependencies=[Depends(get_key_type)])
async def api_usermanager_users_create( async def api_usermanager_users_create(data: CreateUserData):
data: CreateUserData, wallet: WalletTypeInfo = Depends(get_key_type)
):
user = await create_usermanager_user(data) user = await create_usermanager_user(data)
full = user.dict() full = user.dict()
wallets: list[Wallet] = await get_usermanager_users_wallets(user.id)
if wallets:
full["wallets"] = [ full["wallets"] = [
wallet.dict() for wallet in await get_usermanager_users_wallets(user.id) wallet.dict() for wallet in wallets
] ]
return full return full
@usermanager_ext.delete("/api/v1/users/{user_id}") @usermanager_ext.delete("/api/v1/users/{user_id}", dependencies=[Depends(require_admin_key)])
async def api_usermanager_users_delete( async def api_usermanager_users_delete(
user_id, user_id,
delete_core: bool = Query(True), delete_core: bool = Query(True),
wallet: WalletTypeInfo = Depends(require_admin_key),
): ):
user = await get_usermanager_user(user_id) user = await get_usermanager_user(user_id)
if not user: if not user:
@ -84,10 +88,8 @@ async def api_usermanager_activate_extension(
# Wallets # Wallets
@usermanager_ext.post("/api/v1/wallets") @usermanager_ext.post("/api/v1/wallets", dependencies=[Depends(get_key_type)])
async def api_usermanager_wallets_create( async def api_usermanager_wallets_create(data: CreateUserWallet):
data: CreateUserWallet, wallet: WalletTypeInfo = Depends(get_key_type)
):
user = await create_usermanager_wallet( user = await create_usermanager_wallet(
user_id=data.user_id, wallet_name=data.wallet_name, admin_id=data.admin_id user_id=data.user_id, wallet_name=data.wallet_name, admin_id=data.admin_id
) )
@ -95,31 +97,27 @@ async def api_usermanager_wallets_create(
@usermanager_ext.get("/api/v1/wallets") @usermanager_ext.get("/api/v1/wallets")
async def api_usermanager_wallets(wallet: WalletTypeInfo = Depends(require_admin_key)): async def api_usermanager_wallets(
wallet: WalletTypeInfo = Depends(require_admin_key) #type: ignore
):
admin_id = wallet.wallet.user admin_id = wallet.wallet.user
return [wallet.dict() for wallet in await get_usermanager_wallets(admin_id)] return [wallet.dict() for wallet in await get_usermanager_wallets(admin_id)]
@usermanager_ext.get("/api/v1/transactions/{wallet_id}") @usermanager_ext.get("/api/v1/transactions/{wallet_id}", dependencies=[Depends(get_key_type)])
async def api_usermanager_wallet_transactions( async def api_usermanager_wallet_transactions(wallet_id):
wallet_id, wallet: WalletTypeInfo = Depends(get_key_type)
):
return await get_usermanager_wallet_transactions(wallet_id) return await get_usermanager_wallet_transactions(wallet_id)
@usermanager_ext.get("/api/v1/wallets/{user_id}") @usermanager_ext.get("/api/v1/wallets/{user_id}", dependencies=[Depends(require_admin_key)])
async def api_usermanager_users_wallets( async def api_usermanager_users_wallets(user_id):
user_id, wallet: WalletTypeInfo = Depends(require_admin_key)
):
return [ return [
s_wallet.dict() for s_wallet in await get_usermanager_users_wallets(user_id) s_wallet.dict() for s_wallet in await get_usermanager_users_wallets(user_id)
] ]
@usermanager_ext.delete("/api/v1/wallets/{wallet_id}") @usermanager_ext.delete("/api/v1/wallets/{wallet_id}", dependencies=[Depends(require_admin_key)])
async def api_usermanager_wallets_delete( async def api_usermanager_wallets_delete(wallet_id):
wallet_id, wallet: WalletTypeInfo = Depends(require_admin_key)
):
get_wallet = await get_usermanager_wallet(wallet_id) get_wallet = await get_usermanager_wallet(wallet_id)
if not get_wallet: if not get_wallet:
raise HTTPException( raise HTTPException(

View File

@ -110,7 +110,6 @@ exclude = """(?x)(
| ^lnbits/extensions/streamalerts. | ^lnbits/extensions/streamalerts.
| ^lnbits/extensions/tipjar. | ^lnbits/extensions/tipjar.
| ^lnbits/extensions/tpos. | ^lnbits/extensions/tpos.
| ^lnbits/extensions/usermanager.
| ^lnbits/extensions/watchonly. | ^lnbits/extensions/watchonly.
| ^lnbits/extensions/withdraw. | ^lnbits/extensions/withdraw.
| ^lnbits/wallets/lnd_grpc_files. | ^lnbits/wallets/lnd_grpc_files.