5740825225

Committed 02 Aug 2023 03:32PM UTC coverage: 88.958% (+0.05%) from 88.913%

Build # 5740825225

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #844 from geo-engine/pg-symbology-mapping

Pg symbology mapping

Run Details

610 of 610 new or added lines in 10 files covered. (100.0%)

106476 of 119693 relevant lines covered (88.96%)

60487.94 hits per line

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

80.31

/services/src/layers/postgres_layer_db.rs

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

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 metadata = serde_json::to_value(&layer.metadata).context(crate::error::SerdeJson)?;

        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,
                    &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 metadata = serde_json::to_value(&layer.metadata).context(crate::error::SerdeJson)?;

        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,
                    &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: serde_json::from_value(row.get(5)).context(crate::error::SerdeJson)?,
        })
    }
}

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

geo-engine / geoengine / 5740825225

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