5760657609

Committed 04 Aug 2023 08:43AM UTC coverage: 89.424% (+0.002%) from 89.422%

Build # 5760657609

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #847 from geo-engine/pg-layer-metadata-mapping

layer metadata as sql type

Run Details

38 of 38 new or added lines in 4 files covered. (100.0%)

103799 of 116075 relevant lines covered (89.42%)

62375.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.41

/services/src/layers/postgres_layer_db.rs

use crate::api::model::datatypes::{DataProviderId, LayerId};
use crate::api::model::HashMapTextTextDbType;

use crate::contexts::PostgresDb;
use crate::layers::layer::Property;

use crate::{
    error::Result,
    layers::{
        external::{DataProvider, DataProviderDefinition},
        layer::{
            AddLayer, AddLayerCollection, CollectionItem, Layer, LayerCollection,
            LayerCollectionListOptions, LayerCollectionListing, LayerListing,
            ProviderLayerCollectionId, ProviderLayerId,
        },
        listing::{LayerCollectionId, LayerCollectionProvider},
        storage::{
            LayerDb, LayerProviderDb, LayerProviderListing, LayerProviderListingOptions,
            INTERNAL_LAYER_DB_ROOT_COLLECTION_ID, INTERNAL_PROVIDER_ID,
        },
        LayerDbError,
    },
};
use async_trait::async_trait;
use bb8_postgres::tokio_postgres::{
    tls::{MakeTlsConnect, TlsConnect},
    Socket,
};

use snafu::ResultExt;
use std::str::FromStr;
use uuid::Uuid;

/// delete all collections without parent collection
async fn _remove_collections_without_parent_collection(
    transaction: &tokio_postgres::Transaction<'_>,
) -> Result<()> {
    // HINT: a recursive delete statement seems reasonable, but hard to implement in postgres
    //       because you have a graph with potential loops

    let remove_layer_collections_without_parents_stmt = transaction
        .prepare(
            "DELETE FROM layer_collections
                 WHERE  id <> $1 -- do not delete root collection
                 AND    id NOT IN (
                    SELECT child FROM collection_children
                 );",
        )
        .await?;
    while 0 < transaction
        .execute(
            &remove_layer_collections_without_parents_stmt,
            &[&INTERNAL_LAYER_DB_ROOT_COLLECTION_ID],
        )
        .await?
    {
        // whenever one collection is deleted, we have to check again if there are more
        // collections without parents
    }

    Ok(())
}

/// delete all layers without parent collection
async fn _remove_layers_without_parent_collection(
    transaction: &tokio_postgres::Transaction<'_>,
) -> Result<()> {
    let remove_layers_without_parents_stmt = transaction
        .prepare(
            "DELETE FROM layers
                 WHERE id NOT IN (
                    SELECT layer FROM collection_layers
                 );",
        )
        .await?;
    transaction
        .execute(&remove_layers_without_parents_stmt, &[])
        .await?;

    Ok(())
}

#[async_trait]
impl<Tls> LayerDb for PostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_layer(&self, layer: AddLayer, collection: &LayerCollectionId) -> Result<LayerId> {
        let collection_id =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let mut conn = self.conn_pool.get().await?;

        let layer = layer;

        let layer_id = Uuid::new_v4();

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)
            VALUES ($1, $2, $3, $4, $5, $6, $7);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &layer_id,
                    &layer.name,
                    &layer.description,
                    &serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,
                    &layer.symbology,
                    &layer.properties,
                    &HashMapTextTextDbType::from(&layer.metadata),
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "
        INSERT INTO collection_layers (collection, layer)
        VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans.execute(&stmt, &[&collection_id, &layer_id]).await?;

        trans.commit().await?;

        Ok(LayerId(layer_id.to_string()))
    }

    async fn add_layer_with_id(
        &self,
        id: &LayerId,
        layer: AddLayer,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let layer_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let collection_id =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let mut conn = self.conn_pool.get().await?;

        let layer = layer;

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)
            VALUES ($1, $2, $3, $4, $5, $6, $7);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &layer_id,
                    &layer.name,
                    &layer.description,
                    &serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,
                    &layer.symbology,
                    &layer.properties,
                    &HashMapTextTextDbType::from(&layer.metadata),
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO collection_layers (collection, layer)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans.execute(&stmt, &[&collection_id, &layer_id]).await?;

        trans.commit().await?;

        Ok(())
    }

    async fn add_layer_to_collection(
        &self,
        layer: &LayerId,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let layer_id =
            Uuid::from_str(&layer.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: layer.0.clone(),
            })?;

        let collection_id =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
            INSERT INTO collection_layers (collection, layer)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        conn.execute(&stmt, &[&collection_id, &layer_id]).await?;

        Ok(())
    }

    async fn add_layer_collection(
        &self,
        collection: AddLayerCollection,
        parent: &LayerCollectionId,
    ) -> Result<LayerCollectionId> {
        let parent =
            Uuid::from_str(&parent.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: parent.0.clone(),
            })?;

        let mut conn = self.conn_pool.get().await?;

        let collection = collection;

        let collection_id = Uuid::new_v4();

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layer_collections (id, name, description, properties)
            VALUES ($1, $2, $3, $4);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &collection_id,
                    &collection.name,
                    &collection.description,
                    &collection.properties,
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO collection_children (parent, child)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans.execute(&stmt, &[&parent, &collection_id]).await?;

        trans.commit().await?;

        Ok(LayerCollectionId(collection_id.to_string()))
    }

    async fn add_layer_collection_with_id(
        &self,
        id: &LayerCollectionId,
        collection: AddLayerCollection,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let collection_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: id.0.clone(),
            })?;

        let parent =
            Uuid::from_str(&parent.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: parent.0.clone(),
            })?;

        let mut conn = self.conn_pool.get().await?;

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layer_collections (id, name, description, properties)
            VALUES ($1, $2, $3, $4);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &collection_id,
                    &collection.name,
                    &collection.description,
                    &collection.properties,
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO collection_children (parent, child)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans.execute(&stmt, &[&parent, &collection_id]).await?;

        trans.commit().await?;

        Ok(())
    }

    async fn add_collection_to_parent(
        &self,
        collection: &LayerCollectionId,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let collection =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let parent =
            Uuid::from_str(&parent.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: parent.0.clone(),
            })?;

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
            INSERT INTO collection_children (parent, child)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        conn.execute(&stmt, &[&parent, &collection]).await?;

        Ok(())
    }

    async fn remove_layer_collection(&self, collection: &LayerCollectionId) -> Result<()> {
        let collection =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        if collection == INTERNAL_LAYER_DB_ROOT_COLLECTION_ID {
            return Err(LayerDbError::CannotRemoveRootCollection.into());
        }

        let mut conn = self.conn_pool.get().await?;
        let transaction = conn.transaction().await?;

        // delete the collection!
        // on delete cascade removes all entries from `collection_children` and `collection_layers`

        let remove_layer_collection_stmt = transaction
            .prepare(
                "DELETE FROM layer_collections
                 WHERE id = $1;",
            )
            .await?;
        transaction
            .execute(&remove_layer_collection_stmt, &[&collection])
            .await?;

        _remove_collections_without_parent_collection(&transaction).await?;

        _remove_layers_without_parent_collection(&transaction).await?;

        transaction.commit().await.map_err(Into::into)
    }

    async fn remove_layer_from_collection(
        &self,
        layer: &LayerId,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let collection_uuid =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let layer_uuid =
            Uuid::from_str(&layer.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: layer.0.clone(),
            })?;

        let mut conn = self.conn_pool.get().await?;
        let transaction = conn.transaction().await?;

        let remove_layer_collection_stmt = transaction
            .prepare(
                "DELETE FROM collection_layers
                 WHERE collection = $1
                 AND layer = $2;",
            )
            .await?;
        let num_results = transaction
            .execute(
                &remove_layer_collection_stmt,
                &[&collection_uuid, &layer_uuid],
            )
            .await?;

        if num_results == 0 {
            return Err(LayerDbError::NoLayerForGivenIdInCollection {
                layer: layer.clone(),
                collection: collection.clone(),
            }
            .into());
        }

        _remove_layers_without_parent_collection(&transaction).await?;

        transaction.commit().await.map_err(Into::into)
    }

    async fn remove_layer_collection_from_parent(
        &self,
        collection: &LayerCollectionId,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let collection_uuid =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let parent_collection_uuid =
            Uuid::from_str(&parent.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: parent.0.clone(),
            })?;

        let mut conn = self.conn_pool.get().await?;
        let transaction = conn.transaction().await?;

        let remove_layer_collection_stmt = transaction
            .prepare(
                "DELETE FROM collection_children
                 WHERE child = $1
                 AND parent = $2;",
            )
            .await?;
        let num_results = transaction
            .execute(
                &remove_layer_collection_stmt,
                &[&collection_uuid, &parent_collection_uuid],
            )
            .await?;

        if num_results == 0 {
            return Err(LayerDbError::NoCollectionForGivenIdInCollection {
                collection: collection.clone(),
                parent: parent.clone(),
            }
            .into());
        }

        _remove_collections_without_parent_collection(&transaction).await?;

        _remove_layers_without_parent_collection(&transaction).await?;

        transaction.commit().await.map_err(Into::into)
    }
}

#[async_trait]
impl<Tls> LayerCollectionProvider for PostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    #[allow(clippy::too_many_lines)]
    async fn load_layer_collection(
        &self,
        collection_id: &LayerCollectionId,
        options: LayerCollectionListOptions,
    ) -> Result<LayerCollection> {
        let collection = Uuid::from_str(&collection_id.0).map_err(|_| {
            crate::error::Error::IdStringMustBeUuid {
                found: collection_id.0.clone(),
            }
        })?;

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
        SELECT DISTINCT name, description, properties
        FROM layer_collections
        WHERE id = $1;",
            )
            .await?;

        let row = conn.query_one(&stmt, &[&collection]).await?;

        let name: String = row.get(0);
        let description: String = row.get(1);
        let properties: Vec<Property> = row.get(2);

        let stmt = conn
            .prepare(
                "
        SELECT DISTINCT id, name, description, properties, is_layer
        FROM (
            SELECT 
                concat(id, '') AS id, 
                name, 
                description, 
                properties, 
                FALSE AS is_layer
            FROM layer_collections
                JOIN collection_children cc ON (id = cc.child)
            WHERE cc.parent = $1
        ) u UNION (
            SELECT 
                concat(id, '') AS id, 
                name, 
                description, 
                properties, 
                TRUE AS is_layer
            FROM layers uc
                JOIN collection_layers cl ON (id = cl.layer)
            WHERE cl.collection = $1
        )
        ORDER BY is_layer ASC, name ASC
        LIMIT $2 
        OFFSET $3;            
        ",
            )
            .await?;

        let rows = conn
            .query(
                &stmt,
                &[
                    &collection,
                    &i64::from(options.limit),
                    &i64::from(options.offset),
                ],
            )
            .await?;

        let items = rows
            .into_iter()
            .map(|row| {
                let is_layer: bool = row.get(4);

                if is_layer {
                    Ok(CollectionItem::Layer(LayerListing {
                        id: ProviderLayerId {
                            provider_id: INTERNAL_PROVIDER_ID,
                            layer_id: LayerId(row.get(0)),
                        },
                        name: row.get(1),
                        description: row.get(2),
                        properties: row.get(3),
                    }))
                } else {
                    Ok(CollectionItem::Collection(LayerCollectionListing {
                        id: ProviderLayerCollectionId {
                            provider_id: INTERNAL_PROVIDER_ID,
                            collection_id: LayerCollectionId(row.get(0)),
                        },
                        name: row.get(1),
                        description: row.get(2),
                        properties: row.get(3),
                    }))
                }
            })
            .collect::<Result<Vec<CollectionItem>>>()?;

        Ok(LayerCollection {
            id: ProviderLayerCollectionId {
                provider_id: INTERNAL_PROVIDER_ID,
                collection_id: collection_id.clone(),
            },
            name,
            description,
            items,
            entry_label: None,
            properties,
        })
    }

    async fn get_root_layer_collection_id(&self) -> Result<LayerCollectionId> {
        Ok(LayerCollectionId(
            INTERNAL_LAYER_DB_ROOT_COLLECTION_ID.to_string(),
        ))
    }

    async fn load_layer(&self, id: &LayerId) -> Result<Layer> {
        let layer_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: id.0.clone(),
            })?;

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
            SELECT 
                name,
                description,
                workflow,
                symbology,
                properties,
                metadata
            FROM layers
            WHERE id = $1;",
            )
            .await?;

        let row = conn
            .query_one(&stmt, &[&layer_id])
            .await
            .map_err(|_error| LayerDbError::NoLayerForGivenId { id: id.clone() })?;

        Ok(Layer {
            id: ProviderLayerId {
                provider_id: INTERNAL_PROVIDER_ID,
                layer_id: id.clone(),
            },
            name: row.get(0),
            description: row.get(1),
            workflow: serde_json::from_value(row.get(2)).context(crate::error::SerdeJson)?,
            symbology: row.get(3),
            properties: row.get(4),
            metadata: row.get::<_, HashMapTextTextDbType>(5).into(),
        })
    }
}

#[async_trait]
impl<Tls> LayerProviderDb for PostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_layer_provider(
        &self,
        provider: Box<dyn DataProviderDefinition>,
    ) -> Result<DataProviderId> {
        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
              INSERT INTO layer_providers (
                  id, 
                  type_name, 
                  name,
                  definition
              )
              VALUES ($1, $2, $3, $4)",
            )
            .await?;

        let id = provider.id();
        conn.execute(
            &stmt,
            &[
                &id,
                &provider.type_name(),
                &provider.name(),
                &serde_json::to_value(provider)?,
            ],
        )
        .await?;
        Ok(id)
    }

    async fn list_layer_providers(
        &self,
        options: LayerProviderListingOptions,
    ) -> Result<Vec<LayerProviderListing>> {
        // TODO: permission
        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
            SELECT 
                id, 
                name,
                type_name
            FROM 
                layer_providers
            ORDER BY name ASC
            LIMIT $1 
            OFFSET $2;",
            )
            .await?;

        let rows = conn
            .query(
                &stmt,
                &[&i64::from(options.limit), &i64::from(options.offset)],
            )
            .await?;

        Ok(rows
            .iter()
            .map(|row| LayerProviderListing {
                id: row.get(0),
                name: row.get(1),
                description: row.get(2),
            })
            .collect())
    }

    async fn load_layer_provider(&self, id: DataProviderId) -> Result<Box<dyn DataProvider>> {
        // TODO: permissions
        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
               SELECT 
                   definition
               FROM 
                   layer_providers
               WHERE
                   id = $1",
            )
            .await?;

        let row = conn.query_one(&stmt, &[&id]).await?;

        let definition = serde_json::from_value::<Box<dyn DataProviderDefinition>>(row.get(0))?;

        definition.initialize().await
    }
}

1	use crate::api::model::datatypes::{DataProviderId, LayerId};
2	use crate::api::model::HashMapTextTextDbType;
3
4	use crate::contexts::PostgresDb;
5	use crate::layers::layer::Property;
6
7	use crate::{
8	error::Result,
9	layers::{
10	external::{DataProvider, DataProviderDefinition},
11	layer::{
12	AddLayer, AddLayerCollection, CollectionItem, Layer, LayerCollection,
13	LayerCollectionListOptions, LayerCollectionListing, LayerListing,
14	ProviderLayerCollectionId, ProviderLayerId,
15	},
16	listing::{LayerCollectionId, LayerCollectionProvider},
17	storage::{
18	LayerDb, LayerProviderDb, LayerProviderListing, LayerProviderListingOptions,
19	INTERNAL_LAYER_DB_ROOT_COLLECTION_ID, INTERNAL_PROVIDER_ID,
20	},
21	LayerDbError,
22	},
23	};
24	use async_trait::async_trait;
25	use bb8_postgres::tokio_postgres::{
26	tls::{MakeTlsConnect, TlsConnect},
27	Socket,
28	};
29
30	use snafu::ResultExt;
31	use std::str::FromStr;
32	use uuid::Uuid;
33
34	/// delete all collections without parent collection
35	async fn _remove_collections_without_parent_collection(	5✔
36	transaction: &tokio_postgres::Transaction<'_>,	5✔
37	) -> Result<()> {	5✔
38	// HINT: a recursive delete statement seems reasonable, but hard to implement in postgres
39	// because you have a graph with potential loops
40
41	let remove_layer_collections_without_parents_stmt = transaction	5✔
42	.prepare(	5✔
43	"DELETE FROM layer_collections	5✔
44	WHERE id <> $1 -- do not delete root collection	5✔
45	AND id NOT IN (	5✔
46	SELECT child FROM collection_children	5✔
47	);",	5✔
48	)	5✔
49	.await?;	5✔
50	while 0 < transaction	8✔
51	.execute(	8✔
52	&remove_layer_collections_without_parents_stmt,	8✔
53	&[&INTERNAL_LAYER_DB_ROOT_COLLECTION_ID],	8✔
54	)	8✔
55	.await?	8✔
56	{	3✔
57	// whenever one collection is deleted, we have to check again if there are more	3✔
58	// collections without parents	3✔
59	}	3✔
60
61	Ok(())	5✔
62	}	5✔
63
64	/// delete all layers without parent collection
65	async fn _remove_layers_without_parent_collection(	8✔
66	transaction: &tokio_postgres::Transaction<'_>,	8✔
67	) -> Result<()> {	8✔
68	let remove_layers_without_parents_stmt = transaction	8✔
69	.prepare(	8✔
70	"DELETE FROM layers	8✔
71	WHERE id NOT IN (	8✔
72	SELECT layer FROM collection_layers	8✔
73	);",	8✔
74	)	8✔
75	.await?;	8✔
76	transaction	8✔
77	.execute(&remove_layers_without_parents_stmt, &[])	8✔
78	.await?;	8✔
79
80	Ok(())	8✔
81	}	8✔
82
83	#[async_trait]
84	impl<Tls> LayerDb for PostgresDb<Tls>
85	where
86	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
87	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
88	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
89	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
90	{
91	async fn add_layer(&self, layer: AddLayer, collection: &LayerCollectionId) -> Result<LayerId> {	14✔
92	let collection_id =	14✔
93	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	14✔
94	found: collection.0.clone(),	×
95	})?;	14✔
96
97	let mut conn = self.conn_pool.get().await?;	14✔
98
99	let layer = layer;	14✔
100		14✔
101	let layer_id = Uuid::new_v4();	14✔
102
103	let trans = conn.build_transaction().start().await?;	14✔
104
105	let stmt = trans	14✔
106	.prepare(	14✔
107	"	14✔
108	INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)	14✔
109	VALUES ($1, $2, $3, $4, $5, $6, $7);",	14✔
110	)	14✔
111	.await?;	396✔
112
113	trans
114	.execute(
115	&stmt,	14✔
116	&[	14✔
117	&layer_id,	14✔
118	&layer.name,	14✔
119	&layer.description,	14✔
120	&serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,	14✔
121	&layer.symbology,	14✔
122	&layer.properties,	14✔
123	&HashMapTextTextDbType::from(&layer.metadata),	14✔
124	],
125	)
126	.await?;	14✔
127
128	let stmt = trans	14✔
129	.prepare(	14✔
130	"	14✔
131	INSERT INTO collection_layers (collection, layer)	14✔
132	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	14✔
133	)	14✔
134	.await?;	14✔
135
136	trans.execute(&stmt, &[&collection_id, &layer_id]).await?;	14✔
137
138	trans.commit().await?;	15✔
139
140	Ok(LayerId(layer_id.to_string()))	14✔
141	}	28✔
142
143	async fn add_layer_with_id(	×
144	&self,	×
145	id: &LayerId,	×
146	layer: AddLayer,	×
147	collection: &LayerCollectionId,	×
148	) -> Result<()> {	×
149	let layer_id =	×
150	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
151	found: collection.0.clone(),	×
152	})?;	×
153
154	let collection_id =	×
155	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
156	found: collection.0.clone(),	×
157	})?;	×
158
159	let mut conn = self.conn_pool.get().await?;	×
160
161	let layer = layer;	×
162
163	let trans = conn.build_transaction().start().await?;	×
164
165	let stmt = trans	×
166	.prepare(	×
167	"	×
168	INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)	×
169	VALUES ($1, $2, $3, $4, $5, $6, $7);",	×
170	)	×
171	.await?;	×
172
173	trans
174	.execute(
175	&stmt,	×
176	&[	×
177	&layer_id,	×
178	&layer.name,	×
179	&layer.description,	×
180	&serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,	×
181	&layer.symbology,	×
182	&layer.properties,	×
183	&HashMapTextTextDbType::from(&layer.metadata),	×
184	],
185	)
186	.await?;	×
187
188	let stmt = trans	×
189	.prepare(	×
190	"	×
191	INSERT INTO collection_layers (collection, layer)	×
192	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	×
193	)	×
194	.await?;	×
195
196	trans.execute(&stmt, &[&collection_id, &layer_id]).await?;	×
197
198	trans.commit().await?;	×
199
200	Ok(())	×
201	}	×
202
203	async fn add_layer_to_collection(	3✔
204	&self,	3✔
205	layer: &LayerId,	3✔
206	collection: &LayerCollectionId,	3✔
207	) -> Result<()> {	3✔
208	let layer_id =	2✔
209	Uuid::from_str(&layer.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	3✔
210	found: layer.0.clone(),	1✔
211	})?;	3✔
212
213	let collection_id =	2✔
214	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
215	found: collection.0.clone(),	×
216	})?;	2✔
217
218	let conn = self.conn_pool.get().await?;	2✔
219
220	let stmt = conn	2✔
221	.prepare(	2✔
222	"	2✔
223	INSERT INTO collection_layers (collection, layer)	2✔
224	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	2✔
225	)	2✔
226	.await?;	2✔
227
228	conn.execute(&stmt, &[&collection_id, &layer_id]).await?;	2✔
229
230	Ok(())	2✔
231	}	6✔
232
233	async fn add_layer_collection(	20✔
234	&self,	20✔
235	collection: AddLayerCollection,	20✔
236	parent: &LayerCollectionId,	20✔
237	) -> Result<LayerCollectionId> {	20✔
238	let parent =	20✔
239	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	20✔
240	found: parent.0.clone(),	×
241	})?;	20✔
242
243	let mut conn = self.conn_pool.get().await?;	20✔
244
245	let collection = collection;	20✔
246		20✔
247	let collection_id = Uuid::new_v4();	20✔
248
249	let trans = conn.build_transaction().start().await?;	20✔
250
251	let stmt = trans	20✔
252	.prepare(	20✔
253	"	20✔
254	INSERT INTO layer_collections (id, name, description, properties)	20✔
255	VALUES ($1, $2, $3, $4);",	20✔
256	)	20✔
257	.await?;	62✔
258
259	trans	20✔
260	.execute(	20✔
261	&stmt,	20✔
262	&[	20✔
263	&collection_id,	20✔
264	&collection.name,	20✔
265	&collection.description,	20✔
266	&collection.properties,	20✔
267	],	20✔
268	)	20✔
269	.await?;	20✔
270
271	let stmt = trans	20✔
272	.prepare(	20✔
273	"	20✔
274	INSERT INTO collection_children (parent, child)	20✔
275	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	20✔
276	)	20✔
277	.await?;	20✔
278
279	trans.execute(&stmt, &[&parent, &collection_id]).await?;	20✔
280
281	trans.commit().await?;	20✔
282
283	Ok(LayerCollectionId(collection_id.to_string()))	20✔
284	}	40✔
285
286	async fn add_layer_collection_with_id(	×
287	&self,	×
288	id: &LayerCollectionId,	×
289	collection: AddLayerCollection,	×
290	parent: &LayerCollectionId,	×
291	) -> Result<()> {	×
292	let collection_id =	×
293	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
294	found: id.0.clone(),	×
295	})?;	×
296
297	let parent =	×
298	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
299	found: parent.0.clone(),	×
300	})?;	×
301
302	let mut conn = self.conn_pool.get().await?;	×
303
304	let trans = conn.build_transaction().start().await?;	×
305
306	let stmt = trans	×
307	.prepare(	×
308	"	×
309	INSERT INTO layer_collections (id, name, description, properties)	×
310	VALUES ($1, $2, $3, $4);",	×
311	)	×
312	.await?;	×
313
314	trans	×
315	.execute(	×
316	&stmt,	×
317	&[	×
318	&collection_id,	×
319	&collection.name,	×
320	&collection.description,	×
321	&collection.properties,	×
322	],	×
323	)	×
324	.await?;	×
325
326	let stmt = trans	×
327	.prepare(	×
328	"	×
329	INSERT INTO collection_children (parent, child)	×
330	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	×
331	)	×
332	.await?;	×
333
334	trans.execute(&stmt, &[&parent, &collection_id]).await?;	×
335
336	trans.commit().await?;	×
337
338	Ok(())	×
339	}	×
340
341	async fn add_collection_to_parent(	2✔
342	&self,	2✔
343	collection: &LayerCollectionId,	2✔
344	parent: &LayerCollectionId,	2✔
345	) -> Result<()> {	2✔
346	let collection =	2✔
347	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
348	found: collection.0.clone(),	×
349	})?;	2✔
350
351	let parent =	2✔
352	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
353	found: parent.0.clone(),	×
354	})?;	2✔
355
356	let conn = self.conn_pool.get().await?;	2✔
357
358	let stmt = conn	2✔
359	.prepare(	2✔
360	"	2✔
361	INSERT INTO collection_children (parent, child)	2✔
362	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	2✔
363	)	2✔
364	.await?;	2✔
365
366	conn.execute(&stmt, &[&parent, &collection]).await?;	2✔
367
368	Ok(())	2✔
369	}	4✔
370
371	async fn remove_layer_collection(&self, collection: &LayerCollectionId) -> Result<()> {	4✔
372	let collection =	4✔
373	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	4✔
374	found: collection.0.clone(),	×
375	})?;	4✔
376
377	if collection == INTERNAL_LAYER_DB_ROOT_COLLECTION_ID {	4✔
378	return Err(LayerDbError::CannotRemoveRootCollection.into());	1✔
379	}	3✔
380
381	let mut conn = self.conn_pool.get().await?;	3✔
382	let transaction = conn.transaction().await?;	3✔
383
384	// delete the collection!
385	// on delete cascade removes all entries from `collection_children` and `collection_layers`
386
387	let remove_layer_collection_stmt = transaction	3✔
388	.prepare(	3✔
389	"DELETE FROM layer_collections	3✔
390	WHERE id = $1;",	3✔
391	)	3✔
392	.await?;	3✔
393	transaction	3✔
394	.execute(&remove_layer_collection_stmt, &[&collection])	3✔
395	.await?;	3✔
396
397	_remove_collections_without_parent_collection(&transaction).await?;	6✔
398
399	_remove_layers_without_parent_collection(&transaction).await?;	6✔
400
401	transaction.commit().await.map_err(Into::into)	3✔
402	}	8✔
403
404	async fn remove_layer_from_collection(	3✔
405	&self,	3✔
406	layer: &LayerId,	3✔
407	collection: &LayerCollectionId,	3✔
408	) -> Result<()> {	3✔
409	let collection_uuid =	3✔
410	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	3✔
411	found: collection.0.clone(),	×
412	})?;	3✔
413
414	let layer_uuid =	3✔
415	Uuid::from_str(&layer.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	3✔
416	found: layer.0.clone(),	×
417	})?;	3✔
418
419	let mut conn = self.conn_pool.get().await?;	3✔
420	let transaction = conn.transaction().await?;	3✔
421
422	let remove_layer_collection_stmt = transaction	3✔
423	.prepare(	3✔
424	"DELETE FROM collection_layers	3✔
425	WHERE collection = $1	3✔
426	AND layer = $2;",	3✔
427	)	3✔
428	.await?;	3✔
429	let num_results = transaction	3✔
430	.execute(	3✔
431	&remove_layer_collection_stmt,	3✔
432	&[&collection_uuid, &layer_uuid],	3✔
433	)	3✔
434	.await?;	3✔
435
436	if num_results == 0 {	3✔
437	return Err(LayerDbError::NoLayerForGivenIdInCollection {	×
438	layer: layer.clone(),	×
439	collection: collection.clone(),	×
440	}	×
441	.into());	×
442	}	3✔
443		3✔
444	_remove_layers_without_parent_collection(&transaction).await?;	6✔
445
446	transaction.commit().await.map_err(Into::into)	3✔
447	}	6✔
448
449	async fn remove_layer_collection_from_parent(	2✔
450	&self,	2✔
451	collection: &LayerCollectionId,	2✔
452	parent: &LayerCollectionId,	2✔
453	) -> Result<()> {	2✔
454	let collection_uuid =	2✔
455	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
456	found: collection.0.clone(),	×
457	})?;	2✔
458
459	let parent_collection_uuid =	2✔
460	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
461	found: parent.0.clone(),	×
462	})?;	2✔
463
464	let mut conn = self.conn_pool.get().await?;	2✔
465	let transaction = conn.transaction().await?;	2✔
466
467	let remove_layer_collection_stmt = transaction	2✔
468	.prepare(	2✔
469	"DELETE FROM collection_children	2✔
470	WHERE child = $1	2✔
471	AND parent = $2;",	2✔
472	)	2✔
473	.await?;	2✔
474	let num_results = transaction	2✔
475	.execute(	2✔
476	&remove_layer_collection_stmt,	2✔
477	&[&collection_uuid, &parent_collection_uuid],	2✔
478	)	2✔
479	.await?;	2✔
480
481	if num_results == 0 {	2✔
482	return Err(LayerDbError::NoCollectionForGivenIdInCollection {	×
483	collection: collection.clone(),	×
484	parent: parent.clone(),	×
485	}	×
486	.into());	×
487	}	2✔
488		2✔
489	_remove_collections_without_parent_collection(&transaction).await?;	7✔
490
491	_remove_layers_without_parent_collection(&transaction).await?;	4✔
492
493	transaction.commit().await.map_err(Into::into)	2✔
494	}	4✔
495	}
496
497	#[async_trait]
498	impl<Tls> LayerCollectionProvider for PostgresDb<Tls>
499	where
500	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
501	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
502	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
503	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
504	{
505	#[allow(clippy::too_many_lines)]
506	async fn load_layer_collection(	18✔
507	&self,	18✔
508	collection_id: &LayerCollectionId,	18✔
509	options: LayerCollectionListOptions,	18✔
510	) -> Result<LayerCollection> {	18✔
511	let collection = Uuid::from_str(&collection_id.0).map_err(\|_\| {	18✔
512	crate::error::Error::IdStringMustBeUuid {	×
513	found: collection_id.0.clone(),	×
514	}	×
515	})?;	18✔
516
517	let conn = self.conn_pool.get().await?;	18✔
518
519	let stmt = conn	18✔
520	.prepare(	18✔
521	"	18✔
522	SELECT DISTINCT name, description, properties	18✔
523	FROM layer_collections	18✔
524	WHERE id = $1;",	18✔
525	)	18✔
526	.await?;	18✔
527
528	let row = conn.query_one(&stmt, &[&collection]).await?;	18✔
529
530	let name: String = row.get(0);	14✔
531	let description: String = row.get(1);	14✔
532	let properties: Vec<Property> = row.get(2);	14✔
533
534	let stmt = conn	14✔
535	.prepare(	14✔
536	"	14✔
537	SELECT DISTINCT id, name, description, properties, is_layer	14✔
538	FROM (	14✔
539	SELECT	14✔
540	concat(id, '') AS id,	14✔
541	name,	14✔
542	description,	14✔
543	properties,	14✔
544	FALSE AS is_layer	14✔
545	FROM layer_collections	14✔
546	JOIN collection_children cc ON (id = cc.child)	14✔
547	WHERE cc.parent = $1	14✔
548	) u UNION (	14✔
549	SELECT	14✔
550	concat(id, '') AS id,	14✔
551	name,	14✔
552	description,	14✔
553	properties,	14✔
554	TRUE AS is_layer	14✔
555	FROM layers uc	14✔
556	JOIN collection_layers cl ON (id = cl.layer)	14✔
557	WHERE cl.collection = $1	14✔
558	)	14✔
559	ORDER BY is_layer ASC, name ASC	14✔
560	LIMIT $2	14✔
561	OFFSET $3;	14✔
562	",	14✔
563	)	14✔
564	.await?;	14✔
565
566	let rows = conn	14✔
567	.query(	14✔
568	&stmt,	14✔
569	&[	14✔
570	&collection,	14✔
571	&i64::from(options.limit),	14✔
572	&i64::from(options.offset),	14✔
573	],	14✔
574	)	14✔
575	.await?;	14✔
576
577	let items = rows	14✔
578	.into_iter()	14✔
579	.map(\|row\| {	16✔
580	let is_layer: bool = row.get(4);	16✔
581		16✔
582	if is_layer {	16✔
583	Ok(CollectionItem::Layer(LayerListing {	7✔
584	id: ProviderLayerId {	7✔
585	provider_id: INTERNAL_PROVIDER_ID,	7✔
586	layer_id: LayerId(row.get(0)),	7✔
587	},	7✔
588	name: row.get(1),	7✔
589	description: row.get(2),	7✔
590	properties: row.get(3),	7✔
591	}))	7✔
592	} else {
593	Ok(CollectionItem::Collection(LayerCollectionListing {	9✔
594	id: ProviderLayerCollectionId {	9✔
595	provider_id: INTERNAL_PROVIDER_ID,	9✔
596	collection_id: LayerCollectionId(row.get(0)),	9✔
597	},	9✔
598	name: row.get(1),	9✔
599	description: row.get(2),	9✔
600	properties: row.get(3),	9✔
601	}))	9✔
602	}
603	})	16✔
604	.collect::<Result<Vec<CollectionItem>>>()?;	14✔
605
606	Ok(LayerCollection {	14✔
607	id: ProviderLayerCollectionId {	14✔
608	provider_id: INTERNAL_PROVIDER_ID,	14✔
609	collection_id: collection_id.clone(),	14✔
610	},	14✔
611	name,	14✔
612	description,	14✔
613	items,	14✔
614	entry_label: None,	14✔
615	properties,	14✔
616	})	14✔
617	}	36✔
618
619	async fn get_root_layer_collection_id(&self) -> Result<LayerCollectionId> {	17✔
620	Ok(LayerCollectionId(	17✔
621	INTERNAL_LAYER_DB_ROOT_COLLECTION_ID.to_string(),	17✔
622	))	17✔
623	}	17✔
624
625	async fn load_layer(&self, id: &LayerId) -> Result<Layer> {	17✔
626	let layer_id =	17✔
627	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	17✔
628	found: id.0.clone(),	×
629	})?;	17✔
630
631	let conn = self.conn_pool.get().await?;	17✔
632
633	let stmt = conn	17✔
634	.prepare(	17✔
635	"	17✔
636	SELECT	17✔
637	name,	17✔
638	description,	17✔
639	workflow,	17✔
640	symbology,	17✔
641	properties,	17✔
642	metadata	17✔
643	FROM layers	17✔
644	WHERE id = $1;",	17✔
645	)	17✔
646	.await?;	17✔
647
648	let row = conn	17✔
649	.query_one(&stmt, &[&layer_id])	17✔
650	.await	15✔
651	.map_err(\|_error\| LayerDbError::NoLayerForGivenId { id: id.clone() })?;	17✔
652
653	Ok(Layer {
654	id: ProviderLayerId {	13✔
655	provider_id: INTERNAL_PROVIDER_ID,	13✔
656	layer_id: id.clone(),	13✔
657	},	13✔
658	name: row.get(0),	13✔
659	description: row.get(1),	13✔
660	workflow: serde_json::from_value(row.get(2)).context(crate::error::SerdeJson)?,	13✔
661	symbology: row.get(3),	13✔
662	properties: row.get(4),	13✔
663	metadata: row.get::<_, HashMapTextTextDbType>(5).into(),	13✔
664	})
665	}	34✔
666	}
667
668	#[async_trait]
669	impl<Tls> LayerProviderDb for PostgresDb<Tls>
670	where
671	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
672	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
673	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
674	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
675	{
676	async fn add_layer_provider(	5✔
677	&self,	5✔
678	provider: Box<dyn DataProviderDefinition>,	5✔
679	) -> Result<DataProviderId> {	5✔
680	let conn = self.conn_pool.get().await?;	5✔
681
682	let stmt = conn	5✔
683	.prepare(	5✔
684	"	5✔
685	INSERT INTO layer_providers (	5✔
686	id,	5✔
687	type_name,	5✔
688	name,	5✔
689	definition	5✔
690	)	5✔
691	VALUES ($1, $2, $3, $4)",	5✔
692	)	5✔
693	.await?;	5✔
694
695	let id = provider.id();	5✔
696	conn.execute(	5✔
697	&stmt,	5✔
698	&[	5✔
699	&id,	5✔
700	&provider.type_name(),	5✔
701	&provider.name(),	5✔
702	&serde_json::to_value(provider)?,	5✔
703	],
704	)
705	.await?;	5✔
706	Ok(id)	5✔
707	}	10✔
708
709	async fn list_layer_providers(	1✔
710	&self,	1✔
711	options: LayerProviderListingOptions,	1✔
712	) -> Result<Vec<LayerProviderListing>> {	1✔
713	// TODO: permission
714	let conn = self.conn_pool.get().await?;	1✔
715
716	let stmt = conn	1✔
717	.prepare(	1✔
718	"	1✔
719	SELECT	1✔
720	id,	1✔
721	name,	1✔
722	type_name	1✔
723	FROM	1✔
724	layer_providers	1✔
725	ORDER BY name ASC	1✔
726	LIMIT $1	1✔
727	OFFSET $2;",	1✔
728	)	1✔
729	.await?;	1✔
730
731	let rows = conn	1✔
732	.query(	1✔
733	&stmt,	1✔
734	&[&i64::from(options.limit), &i64::from(options.offset)],	1✔
735	)	1✔
736	.await?;	1✔
737
738	Ok(rows	1✔
739	.iter()	1✔
740	.map(\|row\| LayerProviderListing {	1✔
741	id: row.get(0),	1✔
742	name: row.get(1),	1✔
743	description: row.get(2),	1✔
744	})	1✔
745	.collect())	1✔
746	}	2✔
747
748	async fn load_layer_provider(&self, id: DataProviderId) -> Result<Box<dyn DataProvider>> {	12✔
749	// TODO: permissions
750	let conn = self.conn_pool.get().await?;	12✔
751
752	let stmt = conn	12✔
753	.prepare(	12✔
754	"	12✔
755	SELECT	12✔
756	definition	12✔
757	FROM	12✔
758	layer_providers	12✔
759	WHERE	12✔
760	id = $1",	12✔
761	)	12✔
762	.await?;	12✔
763
764	let row = conn.query_one(&stmt, &[&id]).await?;	12✔
765
766	let definition = serde_json::from_value::<Box<dyn DataProviderDefinition>>(row.get(0))?;	7✔
767
768	definition.initialize().await	7✔
769	}	24✔
770	}

geo-engine / geoengine / 5760657609

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous