Helper to return the query for a given geo query.
(field, op, value)
| 419 | |
| 420 | |
| 421 | def _geo_operator(field, op, value): |
| 422 | """Helper to return the query for a given geo query.""" |
| 423 | if op == "max_distance": |
| 424 | value = {"$maxDistance": value} |
| 425 | elif op == "min_distance": |
| 426 | value = {"$minDistance": value} |
| 427 | elif field._geo_index == pymongo.GEO2D: |
| 428 | if op == "within_distance": |
| 429 | value = {"$within": {"$center": value}} |
| 430 | elif op == "within_spherical_distance": |
| 431 | value = {"$within": {"$centerSphere": value}} |
| 432 | elif op == "within_polygon": |
| 433 | value = {"$within": {"$polygon": value}} |
| 434 | elif op == "near": |
| 435 | value = {"$near": value} |
| 436 | elif op == "near_sphere": |
| 437 | value = {"$nearSphere": value} |
| 438 | elif op == "within_box": |
| 439 | value = {"$within": {"$box": value}} |
| 440 | else: |
| 441 | raise NotImplementedError( |
| 442 | 'Geo method "%s" has not been ' "implemented for a GeoPointField" % op |
| 443 | ) |
| 444 | else: |
| 445 | if op == "geo_within": |
| 446 | value = {"$geoWithin": _infer_geometry(value)} |
| 447 | elif op == "geo_within_box": |
| 448 | value = {"$geoWithin": {"$box": value}} |
| 449 | elif op == "geo_within_polygon": |
| 450 | value = {"$geoWithin": {"$polygon": value}} |
| 451 | elif op == "geo_within_center": |
| 452 | value = {"$geoWithin": {"$center": value}} |
| 453 | elif op == "geo_within_sphere": |
| 454 | value = {"$geoWithin": {"$centerSphere": value}} |
| 455 | elif op == "geo_intersects": |
| 456 | value = {"$geoIntersects": _infer_geometry(value)} |
| 457 | elif op == "near": |
| 458 | value = {"$near": _infer_geometry(value)} |
| 459 | else: |
| 460 | raise NotImplementedError( |
| 461 | 'Geo method "{}" has not been implemented for a {} '.format( |
| 462 | op, field._name |
| 463 | ) |
| 464 | ) |
| 465 | return value |
| 466 | |
| 467 | |
| 468 | def _infer_geometry(value): |
no test coverage detected