41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from app.models.peak import Peak
|
|
from app.schemas.peak import PeakCreate, PeakUpdate
|
|
from geoalchemy2.shape import from_shape
|
|
from shapely.geometry import shape, mapping
|
|
import json
|
|
|
|
class CRUDPeak:
|
|
async def get(self, db: AsyncSession, id: int) -> Optional[Peak]:
|
|
result = await db.execute(select(Peak).filter(Peak.id == id))
|
|
return result.scalars().first()
|
|
|
|
async def get_multi(self, db: AsyncSession, skip: int = 0, limit: int = 100) -> List[Peak]:
|
|
result = await db.execute(select(Peak).offset(skip).limit(limit))
|
|
return result.scalars().all()
|
|
|
|
async def create(self, db: AsyncSession, *, obj_in: PeakCreate) -> Peak:
|
|
# Convert GeoJSON dict to WKT for GeoAlchemy2
|
|
# Note: In a real scenario, we might use ST_GeomFromGeoJSON
|
|
# Here assuming obj_in.location is a dict compatible with shapely
|
|
geom_wkt = from_shape(shape(obj_in.location), srid=4326)
|
|
|
|
db_obj = Peak(
|
|
name_en=obj_in.name_en,
|
|
name_cn=obj_in.name_cn,
|
|
elevation_m=obj_in.elevation_m,
|
|
prominence_m=obj_in.prominence_m,
|
|
first_ascent_year=obj_in.first_ascent_year,
|
|
description=obj_in.description,
|
|
thumbnail_url=obj_in.thumbnail_url,
|
|
location=geom_wkt
|
|
)
|
|
db.add(db_obj)
|
|
await db.commit()
|
|
await db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
peak = CRUDPeak()
|