fix: fixed asset. added gcs_helper. added a demo transfer_assets function

Author: jirhiker

.gitignore

@@ -23,7 +23,7 @@ tests/uploads
 migrate.sh
 launcher.sh
 gcs_credentials.json
-
+transfers/data/assets*
 
 # deployment files
 app.yaml

api/asset.py

@@ -24,32 +24,67 @@
 from starlette.status import HTTP_201_CREATED
 
 from api.pagination import CustomPage
-from core.dependencies import session_dependency
+from core.dependencies import session_dependency, viewer_function, admin_dependency, admin_function, editor_dependency
 from db import Thing
 from db.asset import Asset, AssetThingAssociation
 from schemas.asset import AssetResponse, CreateAsset, UpdateAsset
+from services.audit_helper import audit_add
 from services.crud_helper import model_patcher
+from services.gcs_helper import GCS_BUCKET_NAME, get_storage_bucket, gcs_upload, set_asset_url
 
-router = APIRouter(prefix="/asset", tags=["asset"])
-GCS_BUCKET_NAME = os.environ.get("GCS_BUCKET_NAME")
+router = APIRouter(prefix="/asset", tags=["asset"],
+                   dependencies=[Depends(viewer_function)])
 
 
-from google.cloud import storage
+# ======= Create =========
+@router.post("/upload", status_code=HTTP_201_CREATED,
+             dependencies=[Depends(admin_function)])
+async def upload_asset(
+        bucket=Depends(get_storage_bucket), file: UploadFile = File(...)
+):
+    signed_url, blob_name = gcs_upload(bucket, file)
+    return {
+        "url": signed_url,
+        "storage_path": blob_name,
+
+    }
+
+
+@router.post("", status_code=HTTP_201_CREATED)
+async def add_asset(
+        user: admin_dependency,
+        session: session_dependency, asset_data: CreateAsset
+) -> AssetResponse:
+
+    data = asset_data.model_dump()
+    thing_id = data.pop("thing_id", None)
+    url = data.pop("url", "")
 
+    data["storage_service"] = "gcs"
+    asset = Asset(**data)
+    audit_add(user, asset)
 
-def get_storage_bucket() -> storage.Bucket:
-    client = storage.Client.from_service_account_json(
-        os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
-    )
-    bucket = client.bucket(GCS_BUCKET_NAME)
-    return bucket
+    if thing_id:
+        assoc = AssetThingAssociation()
+        audit_add(user, assoc)
+        thing = session.get(Thing, thing_id)
+        assoc.thing = thing
+        assoc.asset = asset
+        session.add(assoc)
+
+    session.add(asset)
+    session.commit()
+    session.refresh(asset)
+    asset.url = url
+    return asset
 
 
+# ======= Read =========
 @router.get("")
 async def list_assets(
     session: session_dependency,
-    # bucket=Depends(get_storage_bucket),
     thing_id: int = None,
+    bucket=Depends(get_storage_bucket)  # Assuming get_storage_bucket is defined elsewhere
 ) -> CustomPage[AssetResponse]:
     """
     List all assets or assets associated with a specific thing.
@@ -60,16 +95,10 @@ async def list_assets(
             AssetThingAssociation.thing_id == thing_id
         )
 
-    # assets = session.scalars(sql).all()
-    # if not assets:
-    #     return []
 
     def transformer(assets: List[Asset]) -> AssetResponse:
-        # blob = bucket.blob(asset.storage_path)
-        # asset.url = blob.generate_signed_url(expiration=timedelta(minutes=10), method="GET")
-        # return [AssetResponse.model_validate(asset) for asset in assets]
         for a in assets:
-            a.url = f"https://storage.googleapis.com/{GCS_BUCKET_NAME}/{a.storage_path}"
+            set_asset_url(a, bucket)
         return assets
 
     return paginate(query=sql, conn=session, transformer=transformer)
@@ -79,9 +108,7 @@ def transformer(assets: List[Asset]) -> AssetResponse:
 async def get_asset(
     asset_id: int,
     session: session_dependency,
-    bucket=Depends(
-        get_storage_bucket
-    ),  # Assuming get_storage_bucket is defined elsewhere
+    bucket=Depends(get_storage_bucket),  # Assuming get_storage_bucket is defined elsewhere
     thing_id: int = None,
 ) -> AssetResponse:
     """
@@ -99,59 +126,21 @@ async def get_asset(
     if not asset:
         raise HTTPException(status_code=404, detail="Asset not found")
 
-    blob = bucket.blob(asset.storage_path)
-    asset.url = blob.generate_signed_url(expiration=timedelta(minutes=10), method="GET")
-    return asset
-
-
-@router.post("/upload", status_code=HTTP_201_CREATED)
-async def upload_asset(
-    bucket=Depends(get_storage_bucket), file: UploadFile = File(...)
-):
-    file_id = str(uuid4())
-    blob_name = f"uploads/{file_id}_{file.filename}"
-    blob = bucket.blob(blob_name)
-    blob.upload_from_file(file.file, content_type=file.content_type)
-    return {
-        "url": blob.generate_signed_url(expiration=timedelta(minutes=10), method="GET"),
-        "storage_path": blob_name,
-    }
-
-
-@router.post("", status_code=HTTP_201_CREATED)
-async def add_asset(
-    session: session_dependency, asset_data: CreateAsset
-) -> AssetResponse:
-
-    data = asset_data.model_dump()
-    thing_id = data.pop("thing_id", None)
-    url = data.pop("url", "")
-
-    data["storage_service"] = "gcs"
-    asset = Asset(**data)
+    set_asset_url(asset, bucket)
 
-    if thing_id:
-        assoc = AssetThingAssociation()
-        thing = session.get(Thing, thing_id)
-        assoc.thing = thing
-        assoc.asset = asset
-        session.add(assoc)
 
-    session.add(asset)
-    session.commit()
-    session.refresh(asset)
-    asset.url = url
     return asset
 
-
+# ======= Update =========
 @router.patch("/{asset_id}")
 async def update_asset(
-    asset_id: int, session: session_dependency, asset_data: UpdateAsset
+    asset_id: int, session: session_dependency, asset_data: UpdateAsset,
+        user: editor_dependency
 ):
     """
     Update an existing asset.
     """
-    return model_patcher(session, Asset, asset_id, asset_data)
+    return model_patcher(session, Asset, asset_id, asset_data, user=user)
 
 
 # ============= EOF =============================================

db/asset.py

@@ -13,8 +13,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ===============================================================================
+from typing import Optional
+
 from sqlalchemy import Column, String, Integer, ForeignKey
-from sqlalchemy.orm import relationship
+from sqlalchemy.orm import relationship, Mapped
+from sqlalchemy.testing.schema import mapped_column
 from sqlalchemy_utils import TSVectorType
 
 from db.base import Base, AutoBaseMixin
@@ -27,14 +30,14 @@ class Asset(Base, AutoBaseMixin):
     # content = Column(UploadedFileField)
     # photo = Column(UploadedFileField(upload_type=UploadedImageWithThumb))
 
-    name = Column(String, nullable=False)
-    label = Column(String, nullable=True)
-    storage_service = Column(String, nullable=False)
-    storage_path = Column(String, nullable=False)
-    mime_type = Column(String, nullable=False)
-    size = Column(Integer, nullable=False)
+    name: Mapped[str]=mapped_column(String, nullable=False)
+    label: Mapped[Optional[str]]=mapped_column(String, nullable=True)
+    storage_service: Mapped[str]=mapped_column(String, nullable=False)
+    storage_path: Mapped[str]=mapped_column(String, nullable=False)
+    mime_type: Mapped[str]=mapped_column(String, nullable=False)
+    size: Mapped[int]=mapped_column(Integer, nullable=False)
 
-    search_vector = Column(
+    search_vector: Mapped[TSVectorType]=mapped_column(
         TSVectorType("name", "mime_type", "storage_service", "storage_path")
     )
 

services/gcs_helper.py

@@ -0,0 +1,55 @@
+# ===============================================================================
+# Copyright 2025 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+import os
+from datetime import timedelta
+from uuid import uuid4
+
+from fastapi import File, UploadFile
+
+GCS_BUCKET_NAME = os.environ.get("GCS_BUCKET_NAME")
+
+
+from google.cloud import storage
+
+
+def get_storage_bucket() -> storage.Bucket:
+    client = storage.Client.from_service_account_json(
+        os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
+    )
+    bucket = client.bucket(GCS_BUCKET_NAME)
+    return bucket
+
+
+def gcs_upload(file: UploadFile, bucket: storage.Bucket=None):
+    if bucket is None:
+        bucket = get_storage_bucket()
+
+    file_id = str(uuid4())
+    head, extension = os.path.splitext(file.filename)
+
+    blob_name = f"uploads/{head}_{file_id}{extension}"
+    blob = bucket.blob(blob_name)
+    blob.upload_from_file(file.file, content_type=file.content_type)
+    signed_url = blob.generate_signed_url(expiration=timedelta(minutes=10), method="GET")
+
+    return signed_url, blob_name
+
+def set_asset_url(asset, bucket=None):
+    if bucket is None:
+        bucket = get_storage_bucket()
+    blob = bucket.blob(asset.storage_path)
+    asset.url = blob.generate_signed_url(expiration=timedelta(minutes=10), method="GET")
+# ============= EOF =============================================

transfers/transfer2.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ===============================================================================
-
+import os
 import time
 import uuid
 from datetime import datetime
@@ -25,6 +25,7 @@
 from shapely import Point
 from shapely.ops import transform
 from sqlalchemy import select
+from starlette.datastructures import UploadFile
 
 from core.app import init_lexicon
 from db import (
@@ -41,10 +42,12 @@
     ThingContactAssociation,
     Base,
     Sensor,
-    Address,
+    Address, Asset, AssetThingAssociation,
 )
 from db.engine import session_ctx
 from schemas.thing import CreateWellScreen
+from services.audit_helper import audit_add
+from services.gcs_helper import gcs_upload
 
 # from db.observation.groundwaterlevel import GroundwaterLevelObservation
 
@@ -457,6 +460,32 @@ def transfer_wellscreens(session, limit=None):
             continue
         # session.add(screen)
 
+def transfer_assets(session):
+    for p in ('asset1.png', 'asset2.png', 'asset3.png'):
+        with open(f"./data/assets/{p}", "rb") as f:
+            uf = UploadFile(file=f,
+                            filename=p,
+                            size=10)
+            url, blob_name = gcs_upload(uf)
+            thing_id = 151
+            asset = Asset(
+                name=p,
+                label=p,
+                storage_path=blob_name,
+                storage_service='gcs',
+                mime_type='image/png',
+                size=uf.size
+            )
+            assoc = AssetThingAssociation()
+            audit_add({'sub': 'foobar',
+                       'name': 'Mr. Foobar'}, assoc)
+            thing = session.get(Thing, thing_id)
+            assoc.thing = thing
+            assoc.asset = asset
+            session.add(assoc)
+            session.add(asset)
+            session.commit()
+
 
 def init_sensor(session):
     sensor = Sensor()
@@ -471,20 +500,20 @@ def init_sensor(session):
 if __name__ == "__main__":
 
     with session_ctx() as sess:
-        Base.metadata.drop_all(sess.bind)
-        Base.metadata.create_all(sess.bind)
-
-        init_lexicon("../core/lexicon.json")
-
-        init_sensor(sess)
-        transfer_wells(sess, 1000)
-        transfer_springs(sess, limit=1000)
-        transfer_perennial_stream(sess)
-        transfer_ephemeral_stream(sess)
-        transfer_met(sess)
-
-        transfer_owners(sess)
-        transfer_wellscreens(sess)
-        transfer_water_levels(sess)
-
+        # Base.metadata.drop_all(sess.bind)
+        # Base.metadata.create_all(sess.bind)
+        #
+        # init_lexicon("../core/lexicon.json")
+        #
+        # init_sensor(sess)
+        # transfer_wells(sess, 1000)
+        # transfer_springs(sess, limit=1000)
+        # transfer_perennial_stream(sess)
+        # transfer_ephemeral_stream(sess)
+        # transfer_met(sess)
+        #
+        # transfer_owners(sess)
+        # transfer_wellscreens(sess)
+        # transfer_water_levels(sess)
+        transfer_assets(sess)
 # ============= EOF =============================================