Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@dt.dtype.register(SQLAlchemyDialect, (ga.Geometry, ga.types._GISType))
def ga_geometry(_, gatype, nullable=True):
t = gatype.geometry_type
if t == 'POINT':
return dt.Point(nullable=nullable)
if t == 'LINESTRING':
return dt.LineString(nullable=nullable)
if t == 'POLYGON':
return dt.Polygon(nullable=nullable)
if t == 'MULTILINESTRING':
return dt.MultiLineString(nullable=nullable)
if t == 'MULTIPOINT':
return dt.MultiPoint(nullable=nullable)
if t == 'MULTIPOLYGON':
return dt.MultiPolygon(nullable=nullable)
if t == 'GEOMETRY':
return dt.Geometry(nullable=nullable)
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column(
"projects",
sa.Column(
"centroid",
geoalchemy2.types.Geometry(geometry_type="POINT", srid=4326),
nullable=True,
),
)
op.add_column(
"projects",
sa.Column(
"geometry",
geoalchemy2.types.Geometry(geometry_type="MULTIPOLYGON", srid=4326),
nullable=True,
),
'__callback__': lambda f: {
'__default_type__': geotypes.Geography if f.geography else geotypes.Geometry,
'__default_type_kwargs__': {
'geometry_type': f.geom_type,
'srid': f.srid,
'dimension': f.dim,
'spatial_index': f.spatial_index,
},
# sqlalchemy.PositiveSmallIntegerField: {'filter_class': NumberFilter},
# sqlalchemy.FloatField: {'filter_class': NumberFilter},
# sqlalchemy.NullBooleanField: {'filter_class': BooleanFilter},
# sqlalchemy.SlugField: {'filter_class': CharFilter},
# sqlalchemy.EmailField: {'filter_class': CharFilter},
# sqlalchemy.FilePathField: {'filter_class': CharFilter},
# sqlalchemy.URLField: {'filter_class': CharFilter},
# sqlalchemy.GenericIPAddressField: {'filter_class': CharFilter},
# sqlalchemy.CommaSeparatedIntegerField: {'filter_class': CharFilter},
# sqlalchemy.UUIDField: {'filter_class': UUIDFilter}
}
FILTER_FOR_DBFIELD_DEFAULT = FILTER_FOR_DBFIELD[sqltypes.VARCHAR]
try:
from geoalchemy2 import types
FILTER_FOR_DBFIELD[types.Geometry] = {'filter_class': WKTFilter, 'lookups': geography_lookups}
FILTER_FOR_DBFIELD[types.Geography] = {'filter_class': WKTFilter, 'lookups': geography_lookups}
except ImportError:
pass
def filter_for_data_type(value):
for date_type, filter_data in FILTER_FOR_DBFIELD.items():
if isinstance(value, date_type):
return filter_data
return FILTER_FOR_DBFIELD_DEFAULT
op.create_index(
"idx_project_info composite",
"project_info",
["locale", "project_id"],
unique=False,
)
op.create_table(
"tasks",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("project_id", sa.Integer(), nullable=False),
sa.Column("x", sa.Integer(), nullable=False),
sa.Column("y", sa.Integer(), nullable=False),
sa.Column("zoom", sa.Integer(), nullable=False),
sa.Column(
"geometry",
geoalchemy2.types.Geometry(geometry_type="MULTIPOLYGON", srid=4326),
nullable=True,
),
sa.Column("task_status", sa.Integer(), nullable=True),
sa.Column("locked_by", sa.BigInteger(), nullable=True),
sa.Column("mapped_by", sa.BigInteger(), nullable=True),
sa.Column("validated_by", sa.BigInteger(), nullable=True),
sa.ForeignKeyConstraint(["locked_by"], ["users.id"], name="fk_users_locked"),
sa.ForeignKeyConstraint(["mapped_by"], ["users.id"], name="fk_users_mapper"),
sa.ForeignKeyConstraint(["project_id"], ["projects.id"]),
sa.ForeignKeyConstraint(
["validated_by"], ["users.id"], name="fk_users_validator"
),
sa.PrimaryKeyConstraint("id", "project_id"),
)
op.create_index(op.f("ix_tasks_project_id"), "tasks", ["project_id"], unique=False)
op.create_table(
'__callback__': lambda f: {
'__default_type__': geotypes.Geography if f.geography else geotypes.Geometry,
'__default_type_kwargs__': {
'geometry_type': f.geom_type,
'srid': f.srid,
'dimension': f.dim,
'spatial_index': f.spatial_index,
},
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"priority_areas",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column(
"geometry",
geoalchemy2.types.Geometry(geometry_type="POLYGON", srid=4326),
nullable=True,
),
sa.PrimaryKeyConstraint("id"),
)
op.create_table(
"project_priority_areas",
sa.Column("project_id", sa.Integer(), nullable=True),
sa.Column("priority_area_id", sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(["priority_area_id"], ["priority_areas.id"]),
sa.ForeignKeyConstraint(["project_id"], ["projects.id"]),
)
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column(
"projects",
sa.Column("aoi_id", sa.INTEGER(), autoincrement=False, nullable=True),
)
op.create_table(
"areas_of_interest",
sa.Column("id", sa.INTEGER(), nullable=False),
sa.Column(
"geometry",
geoalchemy2.types.Geometry(geometry_type="MULTIPOLYGON", srid=4326),
autoincrement=False,
nullable=True,
),
sa.Column(
"centroid",
geoalchemy2.types.Geometry(geometry_type="POINT", srid=4326),
autoincrement=False,
nullable=True,
),
sa.PrimaryKeyConstraint("id", name="areas_of_interest_pkey"),
)
op.create_foreign_key(
"projects_aoi_id_fkey", "projects", "areas_of_interest", ["aoi_id"], ["id"]
)
op.add_column(
"projects",
sa.Column("aoi_id", sa.INTEGER(), autoincrement=False, nullable=True),
)
op.create_table(
"areas_of_interest",
sa.Column("id", sa.INTEGER(), nullable=False),
sa.Column(
"geometry",
geoalchemy2.types.Geometry(geometry_type="MULTIPOLYGON", srid=4326),
autoincrement=False,
nullable=True,
),
sa.Column(
"centroid",
geoalchemy2.types.Geometry(geometry_type="POINT", srid=4326),
autoincrement=False,
nullable=True,
),
sa.PrimaryKeyConstraint("id", name="areas_of_interest_pkey"),
)
op.create_foreign_key(
"projects_aoi_id_fkey", "projects", "areas_of_interest", ["aoi_id"], ["id"]
)
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"areas_of_interest",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column(
"geometry",
geoalchemy2.types.Geometry(geometry_type="MULTIPOLYGON", srid=4326),
nullable=True,
),
sa.Column(
"centroid",
geoalchemy2.types.Geometry(geometry_type="POINT", srid=4326),
nullable=True,
),
sa.PrimaryKeyConstraint("id"),
)
op.create_table(
"tags",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("organisations", sa.String(), nullable=True),
sa.Column("campaigns", sa.String(), nullable=True),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("campaigns"),