blacked

lnbits/extensions/admin/__init__.py

@@ -7,6 +7,7 @@ db = Database("ext_admin")
 
 admin_ext: APIRouter = APIRouter(prefix="/admin", tags=["admin"])
 
+
 def admin_renderer():
     return template_renderer(["lnbits/extensions/admin/templates"])
 

lnbits/extensions/admin/crud.py

@@ -11,13 +11,13 @@ from .models import Admin, Funding
 async def update_wallet_balance(wallet_id: str, amount: int) -> str:
     temp_id = f"temp_{urlsafe_short_hash()}"
     internal_id = f"internal_{urlsafe_short_hash()}"
-    
+
     payment = await create_payment(
         wallet_id=wallet_id,
         checking_id=internal_id,
         payment_request="admin_internal",
         payment_hash="admin_internal",
-        amount=amount*1000,
+        amount=amount * 1000,
         memo="Admin top up",
         pending=False,
     )
@@ -25,6 +25,7 @@ async def update_wallet_balance(wallet_id: str, amount: int) -> str:
     await internal_invoice_queue.put(internal_id)
     return payment
 
+
 async def update_admin(user: str, **kwargs) -> Admin:
     q = ", ".join([f"{field[0]} = ?" for field in kwargs.items()])
     # print("UPDATE", q)
@@ -35,23 +36,37 @@ async def update_admin(user: str, **kwargs) -> Admin:
     assert row, "Newly updated settings couldn't be retrieved"
     return Admin(**row) if row else None
 
+
 async def get_admin() -> Admin:
     row = await db.fetchone("SELECT * FROM admin.admin")
     return Admin(**row) if row else None
 
+
 async def update_funding(data: Funding) -> Funding:
     await db.execute(
         """
         UPDATE admin.funding
         SET backend_wallet = ?, endpoint = ?, port = ?, read_key = ?, invoice_key = ?, admin_key = ?, cert = ?, balance = ?, selected = ?
         WHERE id = ?
-        """, 
-        (data.backend_wallet, data.endpoint, data.port, data.read_key, data.invoice_key, data.admin_key, data.cert, data.balance, data.selected, data.id,),
+        """,
+        (
+            data.backend_wallet,
+            data.endpoint,
+            data.port,
+            data.read_key,
+            data.invoice_key,
+            data.admin_key,
+            data.cert,
+            data.balance,
+            data.selected,
+            data.id,
+        ),
     )
     row = await db.fetchone('SELECT * FROM admin.funding WHERE "id" = ?', (data.id,))
     assert row, "Newly updated settings couldn't be retrieved"
     return Funding(**row) if row else None
 
+
 async def get_funding() -> List[Funding]:
     rows = await db.fetchall("SELECT * FROM admin.funding")
 

lnbits/extensions/admin/migrations.py

@@ -7,19 +7,21 @@ from lnbits.helpers import urlsafe_short_hash
 
 
 async def get_admin_user():
-    if(conf.admin_users[0]):
+    if conf.admin_users[0]:
         return conf.admin_users[0]
     from lnbits.core.crud import create_account, get_user
+
     print("Seems like there's no admin users yet. Let's create an account for you!")
     account = await create_account()
     user = account.id
     assert user, "Newly created user couldn't be retrieved"
-    print(f"Your newly created account/user id is: {user}. This will be the Super Admin user.")
+    print(
+        f"Your newly created account/user id is: {user}. This will be the Super Admin user."
+    )
     conf.admin_users.insert(0, user)
     return user
 
 
-
 async def m001_create_admin_table(db):
     # users/server
     user = await get_admin_user()
@@ -28,7 +30,7 @@ async def m001_create_admin_table(db):
     admin_ext = ",".join(conf.admin_ext)
     disabled_ext = ",".join(conf.disabled_ext)
     funding_source = conf.funding_source
-    #operational
+    # operational
     data_folder = conf.data_folder
     database_url = conf.database_url
     force_https = conf.force_https

lnbits/extensions/admin/models.py

@@ -28,6 +28,7 @@ class UpdateAdminSettings(BaseModel):
     custom_logo: str = Query(None)
     ad_space: str = Query(None)
 
+
 class Admin(BaseModel):
     # users
     user: str
@@ -59,6 +60,7 @@ class Admin(BaseModel):
         data = dict(row)
         return cls(**data)
 
+
 class Funding(BaseModel):
     id: str
     backend_wallet: str

lnbits/extensions/admin/views.py

@@ -16,19 +16,21 @@ from .crud import get_admin, get_funding
 
 templates = Jinja2Templates(directory="templates")
 
+
 @admin_ext.get("/", response_class=HTMLResponse)
 async def index(request: Request, user: User = Depends(check_user_exists)):
     admin = await get_admin()
     funding = [f.dict() for f in await get_funding()]
     error, balance = await g().WALLET.status()
-    
+
     return admin_renderer().TemplateResponse(
-        "admin/index.html", {
+        "admin/index.html",
+        {
             "request": request,
             "user": user.dict(),
             "admin": admin.dict(),
             "funding": funding,
             "settings": g().admin_conf.dict(),
-            "balance": balance
-        }
+            "balance": balance,
+        },
     )

lnbits/extensions/admin/views_api.py

@@ -15,16 +15,18 @@ from .crud import get_admin, update_admin, update_funding, update_wallet_balance
 
 
 @admin_ext.get("/api/v1/admin/{wallet_id}/{topup_amount}", status_code=HTTPStatus.OK)
-async def api_update_balance(wallet_id, topup_amount: int, g: WalletTypeInfo = Depends(require_admin_key)):
+async def api_update_balance(
+    wallet_id, topup_amount: int, g: WalletTypeInfo = Depends(require_admin_key)
+):
     try:
         wallet = await get_wallet(wallet_id)
     except:
         raise HTTPException(
-                status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
-            )
+            status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
+        )
 
     await update_wallet_balance(wallet_id=wallet_id, amount=int(topup_amount))
-    
+
     return {"status": "Success"}
 
 
@@ -32,39 +34,40 @@ async def api_update_balance(wallet_id, topup_amount: int, g: WalletTypeInfo = D
 async def api_update_admin(
     request: Request,
     data: UpdateAdminSettings = Body(...),
-    w: WalletTypeInfo = Depends(require_admin_key)
-    ):
+    w: WalletTypeInfo = Depends(require_admin_key),
+):
     admin = await get_admin()
     # print(data)
     if not admin.user == w.wallet.user:
         raise HTTPException(
-                status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
-            )
+            status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
+        )
     updated = await update_admin(user=w.wallet.user, **data.dict())
 
-    updated.admin_users = removeEmptyString(updated.admin_users.split(','))
-    updated.allowed_users = removeEmptyString(updated.allowed_users.split(','))
-    updated.admin_ext = removeEmptyString(updated.admin_ext.split(','))
-    updated.disabled_ext = removeEmptyString(updated.disabled_ext.split(','))
-    updated.theme = removeEmptyString(updated.theme.split(','))
-    updated.ad_space = removeEmptyString(updated.ad_space.split(','))
+    updated.admin_users = removeEmptyString(updated.admin_users.split(","))
+    updated.allowed_users = removeEmptyString(updated.allowed_users.split(","))
+    updated.admin_ext = removeEmptyString(updated.admin_ext.split(","))
+    updated.disabled_ext = removeEmptyString(updated.disabled_ext.split(","))
+    updated.theme = removeEmptyString(updated.theme.split(","))
+    updated.ad_space = removeEmptyString(updated.ad_space.split(","))
 
     g().admin_conf = g().admin_conf.copy(update=updated.dict())
-    
+
     # print(g().admin_conf)
     return {"status": "Success"}
 
+
 @admin_ext.post("/api/v1/admin/funding/", status_code=HTTPStatus.OK)
 async def api_update_funding(
     request: Request,
     data: Funding = Body(...),
-    w: WalletTypeInfo = Depends(require_admin_key)
-    ):
+    w: WalletTypeInfo = Depends(require_admin_key),
+):
     admin = await get_admin()
 
     if not admin.user == w.wallet.user:
         raise HTTPException(
-                status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
-            )
+            status_code=HTTPStatus.FORBIDDEN, detail="Not allowed: not an admin"
+        )
     funding = await update_funding(data=data)
     return funding