Refactors type hints for clarity
Updates type hints in `crud.py`, `helpers.py`, and `models.py` for improved readability and maintainability. Replaces `Merchant | None` with `Optional[Merchant]` and `list[X]` with `List[X]` for consistency with standard Python typing practices.
This commit is contained in:
parent
429522adba
commit
e8ddc4b697
5 changed files with 122 additions and 115 deletions
69
crud.py
69
crud.py
|
|
@ -1,4 +1,5 @@
|
|||
import json
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from lnbits.helpers import urlsafe_short_hash
|
||||
|
||||
|
|
@ -43,7 +44,7 @@ async def create_merchant(user_id: str, m: PartialMerchant) -> Merchant:
|
|||
|
||||
async def update_merchant(
|
||||
user_id: str, merchant_id: str, config: MerchantConfig
|
||||
) -> Merchant | None:
|
||||
) -> Optional[Merchant]:
|
||||
await db.execute(
|
||||
f"""
|
||||
UPDATE nostrmarket.merchants SET meta = :meta, time = {db.timestamp_now}
|
||||
|
|
@ -54,7 +55,7 @@ async def update_merchant(
|
|||
return await get_merchant(user_id, merchant_id)
|
||||
|
||||
|
||||
async def touch_merchant(user_id: str, merchant_id: str) -> Merchant | None:
|
||||
async def touch_merchant(user_id: str, merchant_id: str) -> Optional[Merchant]:
|
||||
await db.execute(
|
||||
f"""
|
||||
UPDATE nostrmarket.merchants SET time = {db.timestamp_now}
|
||||
|
|
@ -65,7 +66,7 @@ async def touch_merchant(user_id: str, merchant_id: str) -> Merchant | None:
|
|||
return await get_merchant(user_id, merchant_id)
|
||||
|
||||
|
||||
async def get_merchant(user_id: str, merchant_id: str) -> Merchant | None:
|
||||
async def get_merchant(user_id: str, merchant_id: str) -> Optional[Merchant]:
|
||||
row: dict = await db.fetchone(
|
||||
"""SELECT * FROM nostrmarket.merchants WHERE user_id = :user_id AND id = :id""",
|
||||
{
|
||||
|
|
@ -77,7 +78,7 @@ async def get_merchant(user_id: str, merchant_id: str) -> Merchant | None:
|
|||
return Merchant.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_merchant_by_pubkey(public_key: str) -> Merchant | None:
|
||||
async def get_merchant_by_pubkey(public_key: str) -> Optional[Merchant]:
|
||||
row: dict = await db.fetchone(
|
||||
"""SELECT * FROM nostrmarket.merchants WHERE public_key = :public_key""",
|
||||
{"public_key": public_key},
|
||||
|
|
@ -86,7 +87,7 @@ async def get_merchant_by_pubkey(public_key: str) -> Merchant | None:
|
|||
return Merchant.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_merchants_ids_with_pubkeys() -> list[tuple[str, str]]:
|
||||
async def get_merchants_ids_with_pubkeys() -> List[Tuple[str, str]]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"""SELECT id, public_key FROM nostrmarket.merchants""",
|
||||
)
|
||||
|
|
@ -94,7 +95,7 @@ async def get_merchants_ids_with_pubkeys() -> list[tuple[str, str]]:
|
|||
return [(row["id"], row["public_key"]) for row in rows]
|
||||
|
||||
|
||||
async def get_merchant_for_user(user_id: str) -> Merchant | None:
|
||||
async def get_merchant_for_user(user_id: str) -> Optional[Merchant]:
|
||||
row: dict = await db.fetchone(
|
||||
"""SELECT * FROM nostrmarket.merchants WHERE user_id = :user_id """,
|
||||
{"user_id": user_id},
|
||||
|
|
@ -137,7 +138,7 @@ async def create_zone(merchant_id: str, data: Zone) -> Zone:
|
|||
return zone
|
||||
|
||||
|
||||
async def update_zone(merchant_id: str, z: Zone) -> Zone | None:
|
||||
async def update_zone(merchant_id: str, z: Zone) -> Optional[Zone]:
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE nostrmarket.zones
|
||||
|
|
@ -156,7 +157,7 @@ async def update_zone(merchant_id: str, z: Zone) -> Zone | None:
|
|||
return await get_zone(merchant_id, z.id)
|
||||
|
||||
|
||||
async def get_zone(merchant_id: str, zone_id: str) -> Zone | None:
|
||||
async def get_zone(merchant_id: str, zone_id: str) -> Optional[Zone]:
|
||||
row: dict = await db.fetchone(
|
||||
"SELECT * FROM nostrmarket.zones WHERE merchant_id = :merchant_id AND id = :id",
|
||||
{
|
||||
|
|
@ -167,7 +168,7 @@ async def get_zone(merchant_id: str, zone_id: str) -> Zone | None:
|
|||
return Zone.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_zones(merchant_id: str) -> list[Zone]:
|
||||
async def get_zones(merchant_id: str) -> List[Zone]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"SELECT * FROM nostrmarket.zones WHERE merchant_id = :merchant_id",
|
||||
{"merchant_id": merchant_id},
|
||||
|
|
@ -234,7 +235,7 @@ async def create_stall(merchant_id: str, data: Stall) -> Stall:
|
|||
return stall
|
||||
|
||||
|
||||
async def get_stall(merchant_id: str, stall_id: str) -> Stall | None:
|
||||
async def get_stall(merchant_id: str, stall_id: str) -> Optional[Stall]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.stalls
|
||||
|
|
@ -248,7 +249,7 @@ async def get_stall(merchant_id: str, stall_id: str) -> Stall | None:
|
|||
return Stall.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_stalls(merchant_id: str, pending: bool | None = False) -> list[Stall]:
|
||||
async def get_stalls(merchant_id: str, pending: Optional[bool] = False) -> List[Stall]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.stalls
|
||||
|
|
@ -273,7 +274,7 @@ async def get_last_stall_update_time() -> int:
|
|||
return row["event_created_at"] or 0 if row else 0
|
||||
|
||||
|
||||
async def update_stall(merchant_id: str, stall: Stall) -> Stall | None:
|
||||
async def update_stall(merchant_id: str, stall: Stall) -> Optional[Stall]:
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE nostrmarket.stalls
|
||||
|
|
@ -397,7 +398,9 @@ async def update_product(merchant_id: str, product: Product) -> Product:
|
|||
return updated_product
|
||||
|
||||
|
||||
async def update_product_quantity(product_id: str, new_quantity: int) -> Product | None:
|
||||
async def update_product_quantity(
|
||||
product_id: str, new_quantity: int
|
||||
) -> Optional[Product]:
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE nostrmarket.products SET quantity = :quantity
|
||||
|
|
@ -412,7 +415,7 @@ async def update_product_quantity(product_id: str, new_quantity: int) -> Product
|
|||
return Product.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_product(merchant_id: str, product_id: str) -> Product | None:
|
||||
async def get_product(merchant_id: str, product_id: str) -> Optional[Product]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.products
|
||||
|
|
@ -428,8 +431,8 @@ async def get_product(merchant_id: str, product_id: str) -> Product | None:
|
|||
|
||||
|
||||
async def get_products(
|
||||
merchant_id: str, stall_id: str, pending: bool | None = False
|
||||
) -> list[Product]:
|
||||
merchant_id: str, stall_id: str, pending: Optional[bool] = False
|
||||
) -> List[Product]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.products
|
||||
|
|
@ -442,8 +445,8 @@ async def get_products(
|
|||
|
||||
|
||||
async def get_products_by_ids(
|
||||
merchant_id: str, product_ids: list[str]
|
||||
) -> list[Product]:
|
||||
merchant_id: str, product_ids: List[str]
|
||||
) -> List[Product]:
|
||||
# todo: revisit
|
||||
|
||||
keys = []
|
||||
|
|
@ -464,7 +467,7 @@ async def get_products_by_ids(
|
|||
return [Product.from_row(row) for row in rows]
|
||||
|
||||
|
||||
async def get_wallet_for_product(product_id: str) -> str | None:
|
||||
async def get_wallet_for_product(product_id: str) -> Optional[str]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT s.wallet as wallet FROM nostrmarket.products p
|
||||
|
|
@ -571,7 +574,7 @@ async def create_order(merchant_id: str, o: Order) -> Order:
|
|||
return order
|
||||
|
||||
|
||||
async def get_order(merchant_id: str, order_id: str) -> Order | None:
|
||||
async def get_order(merchant_id: str, order_id: str) -> Optional[Order]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.orders
|
||||
|
|
@ -585,7 +588,7 @@ async def get_order(merchant_id: str, order_id: str) -> Order | None:
|
|||
return Order.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_order_by_event_id(merchant_id: str, event_id: str) -> Order | None:
|
||||
async def get_order_by_event_id(merchant_id: str, event_id: str) -> Optional[Order]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.orders
|
||||
|
|
@ -599,7 +602,7 @@ async def get_order_by_event_id(merchant_id: str, event_id: str) -> Order | None
|
|||
return Order.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_orders(merchant_id: str, **kwargs) -> list[Order]:
|
||||
async def get_orders(merchant_id: str, **kwargs) -> List[Order]:
|
||||
q = " AND ".join(
|
||||
[
|
||||
f"{field[0]} = :{field[0]}"
|
||||
|
|
@ -626,7 +629,7 @@ async def get_orders(merchant_id: str, **kwargs) -> list[Order]:
|
|||
|
||||
async def get_orders_for_stall(
|
||||
merchant_id: str, stall_id: str, **kwargs
|
||||
) -> list[Order]:
|
||||
) -> List[Order]:
|
||||
q = " AND ".join(
|
||||
[
|
||||
f"{field[0]} = :{field[0]}"
|
||||
|
|
@ -651,7 +654,7 @@ async def get_orders_for_stall(
|
|||
return [Order.from_row(row) for row in rows]
|
||||
|
||||
|
||||
async def update_order(merchant_id: str, order_id: str, **kwargs) -> Order | None:
|
||||
async def update_order(merchant_id: str, order_id: str, **kwargs) -> Optional[Order]:
|
||||
q = ", ".join(
|
||||
[
|
||||
f"{field[0]} = :{field[0]}"
|
||||
|
|
@ -675,7 +678,7 @@ async def update_order(merchant_id: str, order_id: str, **kwargs) -> Order | Non
|
|||
return await get_order(merchant_id, order_id)
|
||||
|
||||
|
||||
async def update_order_paid_status(order_id: str, paid: bool) -> Order | None:
|
||||
async def update_order_paid_status(order_id: str, paid: bool) -> Optional[Order]:
|
||||
await db.execute(
|
||||
"UPDATE nostrmarket.orders SET paid = :paid WHERE id = :id",
|
||||
{"paid": paid, "id": order_id},
|
||||
|
|
@ -689,7 +692,7 @@ async def update_order_paid_status(order_id: str, paid: bool) -> Order | None:
|
|||
|
||||
async def update_order_shipped_status(
|
||||
merchant_id: str, order_id: str, shipped: bool
|
||||
) -> Order | None:
|
||||
) -> Optional[Order]:
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE nostrmarket.orders
|
||||
|
|
@ -753,7 +756,7 @@ async def create_direct_message(
|
|||
return msg
|
||||
|
||||
|
||||
async def get_direct_message(merchant_id: str, dm_id: str) -> DirectMessage | None:
|
||||
async def get_direct_message(merchant_id: str, dm_id: str) -> Optional[DirectMessage]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.direct_messages
|
||||
|
|
@ -769,7 +772,7 @@ async def get_direct_message(merchant_id: str, dm_id: str) -> DirectMessage | No
|
|||
|
||||
async def get_direct_message_by_event_id(
|
||||
merchant_id: str, event_id: str
|
||||
) -> DirectMessage | None:
|
||||
) -> Optional[DirectMessage]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.direct_messages
|
||||
|
|
@ -783,7 +786,7 @@ async def get_direct_message_by_event_id(
|
|||
return DirectMessage.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_direct_messages(merchant_id: str, public_key: str) -> list[DirectMessage]:
|
||||
async def get_direct_messages(merchant_id: str, public_key: str) -> List[DirectMessage]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.direct_messages
|
||||
|
|
@ -795,7 +798,7 @@ async def get_direct_messages(merchant_id: str, public_key: str) -> list[DirectM
|
|||
return [DirectMessage.from_row(row) for row in rows]
|
||||
|
||||
|
||||
async def get_orders_from_direct_messages(merchant_id: str) -> list[DirectMessage]:
|
||||
async def get_orders_from_direct_messages(merchant_id: str) -> List[DirectMessage]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.direct_messages
|
||||
|
|
@ -856,7 +859,7 @@ async def create_customer(merchant_id: str, data: Customer) -> Customer:
|
|||
return customer
|
||||
|
||||
|
||||
async def get_customer(merchant_id: str, public_key: str) -> Customer | None:
|
||||
async def get_customer(merchant_id: str, public_key: str) -> Optional[Customer]:
|
||||
row: dict = await db.fetchone(
|
||||
"""
|
||||
SELECT * FROM nostrmarket.customers
|
||||
|
|
@ -870,7 +873,7 @@ async def get_customer(merchant_id: str, public_key: str) -> Customer | None:
|
|||
return Customer.from_row(row) if row else None
|
||||
|
||||
|
||||
async def get_customers(merchant_id: str) -> list[Customer]:
|
||||
async def get_customers(merchant_id: str) -> List[Customer]:
|
||||
rows: list[dict] = await db.fetchall(
|
||||
"SELECT * FROM nostrmarket.customers WHERE merchant_id = :merchant_id",
|
||||
{"merchant_id": merchant_id},
|
||||
|
|
@ -878,7 +881,7 @@ async def get_customers(merchant_id: str) -> list[Customer]:
|
|||
return [Customer.from_row(row) for row in rows]
|
||||
|
||||
|
||||
async def get_all_unique_customers() -> list[Customer]:
|
||||
async def get_all_unique_customers() -> List[Customer]:
|
||||
q = """
|
||||
SELECT public_key, MAX(merchant_id) as merchant_id, MAX(event_created_at)
|
||||
FROM nostrmarket.customers
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue