add bulk crud function

.vscode/launch.json

@@ -14,7 +14,7 @@
                 "--host",
                 "0.0.0.0",
                 "--port",
-                "8750",
+                "8751",
                 "--ssl-keyfile=/home/conrad/privkey.pem",
                 "--ssl-certfile=/home/conrad/fullchain.pem",
                 "--reload"

app/api/api_v1/endpoints/joyso.py

@@ -50,10 +50,8 @@ def nft_balence(
 def mint(
     uid: str = "88888888",
     address: str = "0x000000",
-    amount: int = "0",
+    amount: int = "1",
     db: Session = Depends(deps.get_db),
-    skip: int = 0,
-    limit: int = 100,
     current_user: models.users = Depends(deps.get_current_active_superuser),
 ) -> Any:
     """

app/api/api_v1/endpoints/line.py

@@ -43,10 +43,6 @@ class LineRouter(APIRoute):
         return custom_route_handler
 
 
-
-
-
-
 router = APIRouter(route_class=LineRouter)
 
 
@@ -425,14 +421,13 @@ async def nftdrops(userModel: line.NftDrops):
         return "已有資料"
 
 
-
 @router.get("/transactions")
 def transactions(skip: int = 0, limit: int = 100):
     # db connect
     db = dataset.connect(
         'mysql://choozmo:pAssw0rd@db.ptt.cx:3306/arkcard?charset=utf8mb4'
     )
-    sql = 'SELECT * FROM transaction  ' \
+    sql = 'SELECT * FROM transaction ORDER BY id Desc ' \
           'limit '+str(skip)+', '+str(limit)+''
     nft_table = db['nft']
     result = db.query(sql)
@@ -442,9 +437,5 @@ def transactions(skip: int = 0, limit: int = 100):
         row['nft'] = nft_item['title']
         rows.append(row)
     db.close()
-    
-    return rows
-    
-
-
 
+    return rows

app/api/api_v1/endpoints/nft.py

@@ -1,7 +1,7 @@
-from typing import Any, List, Optional
+from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, File, UploadFile, Form
-from pydantic.errors import NotNoneError
+from fastapi.encoders import jsonable_encoder
 from sqlalchemy.orm import Session
 
 from app import crud, models, schemas
@@ -12,7 +12,7 @@ from uuid import uuid4
 router = APIRouter()
 
 
-@router.get("/", response_model=List[schemas.NftPrint])
+@router.get("/", response_model=list[schemas.NftPrint])
 # @router.get("/")
 def read_items(
     db: Session = Depends(deps.get_db),
@@ -24,10 +24,10 @@ def read_items(
     Retrieve items.
     """
     if crud.user.is_superuser(current_user):
-        nfts = crud.nft.get_group_by_title(db, skip=skip, limit=limit)
+        nfts = crud.nft.get_group_by_title(db)
     else:
         nfts = crud.nft.get_multi_by_owner(
-            db=db, owner_id=current_user.userid, skip=skip, limit=limit
+            db=db, owner_id=current_user.userid
         )
     return nfts
 
@@ -50,7 +50,8 @@ async def create_item(
     Create new item.
     """
     if image.content_type.split('/')[0] != 'image':
-        raise HTTPException(status_code=415, detail='content type error! Please upload valid image type')
+        raise HTTPException(status_code=415,
+                            detail='content type error! Please upload valid image type')
     filename = str(uuid4()) + '.' + image.content_type.split('/')[1]
     with open(settings.IMG_PATH + filename, 'wb+') as f:
         f.write(image.file.read())
@@ -68,6 +69,50 @@ async def create_item(
     return nft
 
 
+@router.put("/", response_model=schemas.NftCreate)
+def update_item(
+    *,
+    db: Session = Depends(deps.get_db),
+    title: str,
+    title_new: Optional[str] = Form(None),
+    context: Optional[str] = Form(None),
+    is_active: Optional[bool] = Form(True),
+    catagory: Optional[str] = Form(None),
+    image: Optional[UploadFile] = File(None),
+    current_user: models.users = Depends(deps.get_current_active_user),
+) -> Any:
+    """
+    Update bulk items.
+    """
+    nft = crud.nft.get_by_title(db=db, title=title)
+    if not nft:
+        raise HTTPException(status_code=404, detail="Item not found")
+    # if not crud.user.is_superuser(current_user) and (nft.userid != current_user.userid):
+    if not crud.user.is_superuser(current_user):
+        raise HTTPException(status_code=400, detail="Not enough permissions")
+    item_in = schemas.NftBulkUpdate(**jsonable_encoder(nft[0]))
+    if image:
+        if image.content_type.split('/')[0] != 'image':
+            raise HTTPException(status_code=415,
+                                detail='content type error! Please upload valid image type')
+        filename = str(uuid4()) + '.' + image.content_type.split('/')[1]
+        with open(settings.IMG_PATH + filename, 'wb+') as f:
+            f.write(image.file.read())
+            f.close()
+        item_in.imgurl = settings.IMG_HOST + filename
+    if title_new:
+        item_in.title = title_new
+    if context:
+        item_in.context = context
+    if is_active:
+        item_in.is_active = is_active
+    if catagory:
+        item_in.category = catagory
+    
+    nft = crud.nft.update_bulk_title(db=db, db_obj_list=nft, obj_in=item_in)
+    return nft
+
+
 @router.put("/{id}", response_model=schemas.NftCreate)
 def update_item(
     *,

app/crud/crud_nft.py

@@ -1,15 +1,30 @@
-from typing import List
+from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
+from fastapi.encoders import jsonable_encoder
 
 from sqlalchemy.orm import Session
 from sqlalchemy import func
-from sqlalchemy.sql.elements import Null
+from itertools import groupby
+from operator import attrgetter
+from pydantic import BaseModel
+from app.db.base_class import Base
+
+
 
 from app.crud.base import CRUDBase
 from app.models.nft import Nft
 from app.schemas.nft import NftBase, NftCreate, NftUpdate # noqa
 
+ModelType = TypeVar("ModelType", bound=Base)
+CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
+UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
+
 
 class CRUDNft(CRUDBase[Nft, NftBase, NftCreate]):
+    def get_by_title(
+        self, db: Session, *, title: str
+    ) -> list[Nft]:
+        return db.query(Nft).filter(Nft.title == title).all()
+
     def get_multi_by_owner(
         self, db: Session, *, skip: int = 0, limit: int = 100, owner_id: str
     ) -> List[Nft]:
@@ -19,6 +34,10 @@ class CRUDNft(CRUDBase[Nft, NftBase, NftCreate]):
     def get_group_by_title(
         self, db: Session, *, skip: int = 0, limit: int = 100
     ) -> List[Nft]:
+        # nft_list = db.query(Nft).order_by(Nft.title).all()
+        # return {k: {"Nft": list(g), "count": db.query(Nft.title, func.count(Nft.userid)
+        #         .label('count')).group_by(Nft.title).filter(Nft.title == k).first()[1]}
+        #         for k, g in groupby(nft_list, attrgetter('title'))}
         return db.query(Nft, func.count(Nft.userid).label('count'))\
             .group_by(Nft.title).offset(skip).limit(limit).all()
 
@@ -45,5 +64,23 @@ class CRUDNft(CRUDBase[Nft, NftBase, NftCreate]):
         db.refresh(db_obj)
         return db_obj
 
+    def update_bulk_title(
+        self, db: Session, *, db_obj_list: list[NftUpdate],
+        obj_in: NftUpdate
+    ) -> Nft:
+        for db_obj in db_obj_list:
+            obj_data = jsonable_encoder(db_obj)
+            if isinstance(obj_in, dict):
+                update_data = obj_in
+            else:
+                update_data = obj_in.dict(exclude_unset=True)
+            for field in obj_data:
+                if field in update_data:
+                    setattr(db_obj, field, update_data[field])
+            db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
 
 nft = CRUDNft(Nft)

app/schemas/nft.py

@@ -7,10 +7,11 @@ class NftBase(BaseModel):
     hash: Optional[str] = None
     imgurl: Optional[str] = None
     userid: Optional[str] = None
-    title: Optional[str] = None
+    title: str
     context: Optional[str] = None
     is_active: Optional[bool] = True
     category: Optional[str] = None
+    # id: int
 
     class Config:
         orm_mode = True
@@ -30,3 +31,11 @@ class NftCreate(NftBase):
 
 class NftUpdate(NftBase):
     pass
+
+
+class NftBulkUpdate(BaseModel):
+    imgurl: Optional[str] = None
+    title: str
+    context: Optional[str] = None
+    is_active: Optional[bool] = True
+    category: Optional[str] = None