unitycatalog_server/api/
functions.rs

1use itertools::Itertools;
2
3use unitycatalog_common::models::functions::v1::*;
4use unitycatalog_common::models::{ObjectLabel, ResourceIdent, ResourceName};
5
6use super::{RequestContext, SecuredAction};
7use crate::Result;
8pub use crate::codegen::functions::FunctionHandler;
9use crate::policy::{Permission, Policy, process_resources};
10use crate::store::ResourceStore;
11
12#[async_trait::async_trait]
13impl<T: ResourceStore + Policy<RequestContext>> FunctionHandler<RequestContext> for T {
14    #[tracing::instrument(skip(self, context), fields(resource_name))]
15    async fn list_functions(
16        &self,
17        request: ListFunctionsRequest,
18        context: RequestContext,
19    ) -> Result<ListFunctionsResponse> {
20        self.check_required(&request, &context).await?;
21        let (mut resources, next_page_token) = self
22            .list(
23                &ObjectLabel::Function,
24                Some(&ResourceName::new([
25                    &request.catalog_name,
26                    &request.schema_name,
27                ])),
28                super::page_size(request.max_results)?,
29                request.page_token,
30            )
31            .await?;
32        process_resources(self, &context, &Permission::Read, &mut resources).await?;
33        Ok(ListFunctionsResponse {
34            functions: resources.into_iter().map(|r| r.try_into()).try_collect()?,
35            next_page_token,
36            ..Default::default()
37        })
38    }
39
40    #[tracing::instrument(skip(self, context), fields(resource_name))]
41    async fn create_function(
42        &self,
43        request: CreateFunctionRequest,
44        context: RequestContext,
45    ) -> Result<Function> {
46        self.check_required(&request, &context).await?;
47        let function_info = request
48            .function_info
49            .ok_or_else(|| crate::Error::invalid_argument("function_info must be provided"))?;
50        tracing::Span::current().record("resource_name", &function_info.name);
51        let full_name = format!(
52            "{}.{}.{}",
53            function_info.catalog_name, function_info.schema_name, function_info.name
54        );
55        let resource = Function {
56            name: function_info.name,
57            catalog_name: function_info.catalog_name,
58            schema_name: function_info.schema_name,
59            full_name,
60            data_type: function_info.data_type,
61            full_data_type: function_info.full_data_type,
62            input_params: Some(function_info.input_params.into_option().unwrap_or_default()).into(),
63            parameter_style: function_info.parameter_style,
64            is_deterministic: function_info.is_deterministic,
65            sql_data_access: function_info.sql_data_access,
66            is_null_call: function_info.is_null_call,
67            security_type: function_info.security_type,
68            routine_body: function_info.routine_body,
69            routine_definition: function_info.routine_definition,
70            external_language: function_info.external_language,
71            comment: function_info.comment,
72            properties: function_info.properties,
73            return_params: function_info.return_params,
74            routine_dependencies: function_info.routine_dependencies,
75            specific_name: function_info.specific_name,
76            ..Default::default()
77        };
78        Ok(self.create(resource.into()).await?.0.try_into()?)
79    }
80
81    #[tracing::instrument(skip(self, context), fields(resource_name))]
82    async fn get_function(
83        &self,
84        request: GetFunctionRequest,
85        context: RequestContext,
86    ) -> Result<Function> {
87        tracing::Span::current().record("resource_name", &request.name);
88        self.check_required(&request, &context).await?;
89        Ok(self.get(&request.resource()).await?.0.try_into()?)
90    }
91
92    #[tracing::instrument(skip(self, context), fields(resource_name))]
93    async fn update_function(
94        &self,
95        request: UpdateFunctionRequest,
96        context: RequestContext,
97    ) -> Result<Function> {
98        tracing::Span::current().record("resource_name", &request.name);
99        self.check_required(&request, &context).await?;
100        let ident = request.resource();
101        let resource = Function {
102            owner: request.owner,
103            ..Default::default()
104        };
105        Ok(self.update(&ident, resource.into()).await?.0.try_into()?)
106    }
107
108    #[tracing::instrument(skip(self, context), fields(resource_name))]
109    async fn delete_function(
110        &self,
111        request: DeleteFunctionRequest,
112        context: RequestContext,
113    ) -> Result<()> {
114        tracing::Span::current().record("resource_name", &request.name);
115        self.check_required(&request, &context).await?;
116        Ok(self.delete(&request.resource()).await?)
117    }
118}
119
120impl SecuredAction for CreateFunctionRequest {
121    fn resource(&self) -> ResourceIdent {
122        let info = self.function_info.as_option();
123        ResourceIdent::function(ResourceName::new([
124            info.map(|i| i.catalog_name.as_str()).unwrap_or_default(),
125            info.map(|i| i.schema_name.as_str()).unwrap_or_default(),
126            info.map(|i| i.name.as_str()).unwrap_or_default(),
127        ]))
128    }
129
130    fn permission(&self) -> &'static Permission {
131        &Permission::Create
132    }
133}
134
135impl SecuredAction for ListFunctionsRequest {
136    fn resource(&self) -> ResourceIdent {
137        ResourceIdent::function(ResourceName::new([
138            self.catalog_name.as_str(),
139            self.schema_name.as_str(),
140        ]))
141    }
142
143    fn permission(&self) -> &'static Permission {
144        &Permission::Read
145    }
146}
147
148impl SecuredAction for GetFunctionRequest {
149    fn resource(&self) -> ResourceIdent {
150        ResourceIdent::function(ResourceName::from_naive_str_split(self.name.as_str()))
151    }
152
153    fn permission(&self) -> &'static Permission {
154        &Permission::Read
155    }
156}
157
158impl SecuredAction for UpdateFunctionRequest {
159    fn resource(&self) -> ResourceIdent {
160        ResourceIdent::function(ResourceName::from_naive_str_split(self.name.as_str()))
161    }
162
163    fn permission(&self) -> &'static Permission {
164        &Permission::Manage
165    }
166}
167
168impl SecuredAction for DeleteFunctionRequest {
169    fn resource(&self) -> ResourceIdent {
170        ResourceIdent::function(ResourceName::from_naive_str_split(self.name.as_str()))
171    }
172
173    fn permission(&self) -> &'static Permission {
174        &Permission::Manage
175    }
176}
unitycatalog_server/api/functions.rs

unitycatalog_server/api/
functions.rs
