I need to insert a point into Column(Geometry(geometry_type='POINT', srid=4326))
. I can not figure out how to implement the srid=4326
into the point data POINT(58.183594 22.593726)
. I searched for it. I still cannot solve this!
Error:
sqlalchemy.exc.DataError: (psycopg2.errors.InvalidParameterValue) Geometry SRID (0) does not match column SRID (4326)
[SQL: INSERT INTO addresses (label, coordinates, building_number, flat_number, disabled, created_at, updated_at, user_id) VALUES (%(label)s, ST_GeomFromEWKT(%(coordinates)s), %(building_number)s, %(flat_number)s, %(disabled)s, %(created_at)s, %(updated_at)s, %(user_id)s) RETURNING addresses.id]
[parameters: {'label': 'my home', 'coordinates': 'POINT(58.183594 22.593726)', 'building_number': 1, 'flat_number': None, 'disabled': False, 'created_at': datetime.datetime(2020, 9, 28, 11, 11, 3, 913693), 'updated_at': datetime.datetime(2020, 9, 28, 11, 11, 3, 913706), 'user_id': 1}]
FastAPI route:
@router.post("/{id}/addresses", response_model=schemas.AddressOut)
async def create_a_new_address(id:int, address_in: schemas.AddressIn, db: Session = Depends(get_db), token_data: schemas.TokenDecoded = Depends(auth.decode_token)):
if id == token_data.id or auth.can_access(token_data.bitrole, "USERS"):
address_in_db= schemas.AddressInDB(**address_in.dict(), user_id=id)
return crud.create_address(db=db, address=address_in_db)
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="access denied"
)
pydantic (schemas.py):
def set_now(time: datetime.datetime)-> datetime.datetime:
return time or datetime.datetime.now()
def tuple_to_point(coord_tuple: Tuple)->str:
return f'POINT({coord_tuple[0]} {coord_tuple[1]})'
class AddressIn(BaseModel):
label: str = ...
coordinates: Tuple[float,float] = ...
building_number: int = ...
flat_number: Optional[int]=None
class AddressInDB(AddressIn):
created_at: datetime.datetime = None
updated_at: datetime.datetime = None
coordinates: str = ...
user_id: int = ...
_created_at = validator('created_at', allow_reuse=True, pre=True, always=True)(set_now)
_updated_at = validator('updated_at', allow_reuse=True, pre=True, always=True)(set_now)
_coordinates = validator('coordinates', allow_reuse=True, pre=True, always=True)(tuple_to_point)
class AddressOut(AddressInDB):
id: int
sqlalchemy,geoalchemy2 (tables.py):
class Addresses(Base):
__tablename__="addresses"
id= Column(Integer, primary_key=True, index=True)
label= Column(String)
coordinates= Column(Geometry(geometry_type='POINT', srid=4326))
building_number= Column(String)
flat_number= Column(String, nullable=True)
disabled = Column(Boolean, default=False)
created_at= Column(DateTime)
updated_at= Column(DateTime)
user_id = Column(Integer, ForeignKey('users.id'))
crud.py
def create_address(db: Session, address: schemas.AddressInDB):
db_address = tables.Addresses(**address.dict())
db.add(db_address)
try:
db.commit()
except IntegrityError as e:
process_db_error(e)
db.refresh(db_address)
return db_address
Use the EWKT representation: