unitycatalog_server/api/
shares.rs

1use itertools::Itertools;
2use std::collections::HashMap;
3
4use unitycatalog_common::models::ObjectLabel;
5use unitycatalog_common::models::shares::v1::*;
6use unitycatalog_common::models::{ResourceIdent, ResourceName, ResourceRef};
7
8use super::{RequestContext, SecuredAction};
9pub use crate::codegen::shares::ShareHandler;
10use crate::policy::{Permission, Policy, process_resources};
11use crate::store::ResourceStore;
12use crate::{Error, Result};
13
14#[async_trait::async_trait]
15impl<T: ResourceStore + Policy<RequestContext>> ShareHandler<RequestContext> for T {
16    #[tracing::instrument(skip(self, context), fields(resource_name))]
17    async fn create_share(
18        &self,
19        request: CreateShareRequest,
20        context: RequestContext,
21    ) -> Result<Share> {
22        tracing::Span::current().record("resource_name", &request.name);
23        self.check_required(&request, &context).await?;
24        let resource = Share {
25            name: request.name,
26            comment: request.comment,
27            ..Default::default()
28        };
29        // TODO:
30        // - update the share with the current actor as owner
31        // - create updated_* relations
32        Ok(self.create(resource.into()).await?.0.try_into()?)
33    }
34
35    #[tracing::instrument(skip(self, context), fields(resource_name))]
36    async fn delete_share(
37        &self,
38        request: DeleteShareRequest,
39        context: RequestContext,
40    ) -> Result<()> {
41        tracing::Span::current().record("resource_name", &request.name);
42        self.check_required(&request, &context).await?;
43        Ok(self.delete(&request.resource()).await?)
44    }
45
46    #[tracing::instrument(skip(self, context))]
47    async fn list_shares(
48        &self,
49        request: ListSharesRequest,
50        context: RequestContext,
51    ) -> Result<ListSharesResponse> {
52        self.check_required(&request, &context).await?;
53        let (mut resources, next_page_token) = self
54            .list(
55                &ObjectLabel::Share,
56                None,
57                request.max_results.map(|v| v as usize),
58                request.page_token,
59            )
60            .await?;
61        process_resources(self, &context, &Permission::Read, &mut resources).await?;
62        Ok(ListSharesResponse {
63            shares: resources.into_iter().map(|r| r.try_into()).try_collect()?,
64            next_page_token,
65            ..Default::default()
66        })
67    }
68
69    #[tracing::instrument(skip(self, context), fields(resource_name))]
70    async fn get_share(&self, request: GetShareRequest, context: RequestContext) -> Result<Share> {
71        tracing::Span::current().record("resource_name", &request.name);
72        self.check_required(&request, &context).await?;
73        Ok(self.get(&request.resource()).await?.0.try_into()?)
74    }
75
76    #[tracing::instrument(skip(self, context), fields(resource_name))]
77    async fn update_share(
78        &self,
79        request: UpdateShareRequest,
80        context: RequestContext,
81    ) -> Result<Share> {
82        tracing::Span::current().record("resource_name", &request.name);
83        self.check_required(&request, &context).await?;
84        let ident = request.resource();
85        let current: Share = self.get(&ident).await?.0.try_into()?;
86
87        // update the data_objects according to the actions defined in request
88        let mut data_objects: HashMap<String, DataObject> = current
89            .objects
90            .into_iter()
91            .map(|d| (d.name.clone(), d))
92            .collect();
93        for update in request.updates.iter() {
94            match update.action.as_known().unwrap_or_default() {
95                Action::ADD => {
96                    if let Some(obj) = update.data_object.as_option() {
97                        if data_objects.contains_key(&obj.name) {
98                            return Err(Error::AlreadyExists);
99                        }
100                        data_objects.insert(obj.name.clone(), obj.clone());
101                    }
102                }
103                Action::REMOVE => {
104                    if let Some(obj) = update.data_object.as_option() {
105                        data_objects.remove(&obj.name);
106                    }
107                }
108                Action::UPDATE => {
109                    if let Some(obj) = update.data_object.as_option() {
110                        if let Some(existing) = data_objects.get_mut(&obj.name) {
111                            *existing = obj.clone();
112                        } else {
113                            return Err(Error::NotFound);
114                        }
115                    }
116                }
117                Action::ACTION_UNSPECIFIED => {
118                    return Err(Error::InvalidArgument("Unspecified action".to_string()));
119                }
120            }
121        }
122
123        let resource = Share {
124            name: request.new_name.unwrap_or_else(|| request.name.clone()),
125            comment: request.comment.or(current.comment),
126            owner: request.owner.or(current.owner),
127            objects: data_objects.into_values().collect(),
128            ..Default::default()
129        };
130        // TODO:
131        // - add update_* relations
132        Ok(self.update(&ident, resource.into()).await?.0.try_into()?)
133    }
134
135    #[tracing::instrument(skip(self, context), fields(resource_name))]
136    async fn get_permissions(
137        &self,
138        request: GetPermissionsRequest,
139        context: RequestContext,
140    ) -> Result<GetPermissionsResponse> {
141        tracing::Span::current().record("resource_name", &request.name);
142        self.check_required(&request, &context).await?;
143        // Permissions for a share are modelled as associations between the share and
144        // principal-named resources.  The `AssociationLabel` enum currently does not
145        // include a dedicated `SharedWith` / `GrantedTo` variant, so we cannot store
146        // per-principal privilege lists in the association graph today.
147        //
148        // Once a suitable `AssociationLabel` variant is added (e.g. `GrantedTo` /
149        // `GrantedBy`) the implementation should:
150        //   1. Call `self.list_associations(&share_ident, &AssociationLabel::GrantedTo, …)`
151        //   2. For each returned ResourceIdent, read the stored privilege list from the
152        //      association properties and build a `PrivilegeAssignment`.
153        //
154        // For now we return an empty list so that callers receive a valid (non-panicking)
155        // response and the endpoint is reachable.
156        let _ = self.get(&request.resource()).await?; // verify the share exists
157        Ok(GetPermissionsResponse {
158            privilege_assignments: vec![],
159            next_page_token: None,
160            ..Default::default()
161        })
162    }
163
164    #[tracing::instrument(skip(self, context), fields(resource_name))]
165    async fn update_permissions(
166        &self,
167        request: UpdatePermissionsRequest,
168        context: RequestContext,
169    ) -> Result<UpdatePermissionsResponse> {
170        tracing::Span::current().record("resource_name", &request.name);
171        self.check_required(&request, &context).await?;
172        // See the note in `get_permissions`: a dedicated `AssociationLabel` variant is
173        // needed before per-principal privilege changes can be persisted in the graph.
174        //
175        // Once that label exists, the implementation should iterate `request.changes` and:
176        //   - For adds:    `self.add_association(&share_ident, &principal_ident, &label, props)`
177        //   - For removes: `self.remove_association(&share_ident, &principal_ident, &label)`
178        //
179        // For now we verify the share exists, acknowledge the request, and return the
180        // current (empty) privilege list unless the caller asked to omit it.
181        let _ = self.get(&request.resource()).await?; // verify the share exists
182        // When omit_permissions_list is true the caller does not want the updated list back.
183        // Both cases currently return an empty list; once the association label is added the
184        // non-omit path should call get_permissions to build the real list.
185        Ok(UpdatePermissionsResponse {
186            privilege_assignments: vec![],
187            ..Default::default()
188        })
189    }
190}
191
192impl SecuredAction for CreateShareRequest {
193    fn resource(&self) -> ResourceIdent {
194        ResourceIdent::share(ResourceName::new([self.name.as_str()]))
195    }
196
197    fn permission(&self) -> &'static Permission {
198        &Permission::Create
199    }
200}
201
202impl SecuredAction for ListSharesRequest {
203    fn resource(&self) -> ResourceIdent {
204        ResourceIdent::share(ResourceRef::Undefined)
205    }
206
207    fn permission(&self) -> &'static Permission {
208        &Permission::Read
209    }
210}
211
212impl SecuredAction for GetShareRequest {
213    fn resource(&self) -> ResourceIdent {
214        ResourceIdent::share(ResourceName::new([self.name.as_str()]))
215    }
216
217    fn permission(&self) -> &'static Permission {
218        &Permission::Read
219    }
220}
221
222impl SecuredAction for UpdateShareRequest {
223    fn resource(&self) -> ResourceIdent {
224        ResourceIdent::share(ResourceName::new([self.name.as_str()]))
225    }
226
227    fn permission(&self) -> &'static Permission {
228        &Permission::Manage
229    }
230}
231
232impl SecuredAction for DeleteShareRequest {
233    fn resource(&self) -> ResourceIdent {
234        ResourceIdent::share(ResourceName::new([self.name.as_str()]))
235    }
236
237    fn permission(&self) -> &'static Permission {
238        &Permission::Manage
239    }
240}
241
242impl SecuredAction for GetPermissionsRequest {
243    fn resource(&self) -> ResourceIdent {
244        ResourceIdent::share(ResourceName::new([self.name.as_str()]))
245    }
246
247    fn permission(&self) -> &'static Permission {
248        &Permission::Read
249    }
250}
251
252impl SecuredAction for UpdatePermissionsRequest {
253    fn resource(&self) -> ResourceIdent {
254        ResourceIdent::share(ResourceName::new([self.name.as_str()]))
255    }
256
257    fn permission(&self) -> &'static Permission {
258        &Permission::Manage
259    }
260}
unitycatalog_server/api/shares.rs

unitycatalog_server/api/
shares.rs
