Skip to content

Commit 3dac0f2

Browse files
rmn-boikormn-boiko
authored
Add wildcard for pull and push (#506)
1 parent 1e5ed26 commit 3dac0f2

16 files changed

Lines changed: 721 additions & 86 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
44
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
55
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
66

7+
## [Unreleased]
8+
### Added
9+
- Added support of wildcard patterns for `kamu pull` and `kamu push` commands
10+
711
## [0.163.1] - 2024-03-01
812
### Fixed
913
- Fixed `mapboxgl` dependency issue in Jupyter image

‎resources/cli-reference.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -603,6 +603,10 @@ Fetch latest data in a specific dataset:
603603

604604
kamu pull org.example.data
605605

606+
Fetch latest data in datasets matching pattern:
607+
608+
kamu pull org.example.%
609+
606610
Fetch latest data for the entire dependency tree of a dataset:
607611

608612
kamu pull --recursive org.example.derivative
@@ -664,6 +668,10 @@ Sync dataset that already has a push alias:
664668

665669
kamu push org.example.data
666670

671+
Sync datasets matching pattern that already have push aliases:
672+
673+
kamu push org.example.%
674+
667675
Add dataset to local IPFS node and update IPNS entry to the new CID:
668676

669677
kamu push org.example.data --to ipns://k5..zy

‎src/app/cli/src/cli_commands.rs‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -262,6 +262,7 @@ pub fn get_command(
262262
cli_catalog.get_one()?,
263263
cli_catalog.get_one()?,
264264
cli_catalog.get_one()?,
265+
cli_catalog.get_one()?,
265266
datasets,
266267
submatches.get_flag("all"),
267268
submatches.get_flag("recursive"),
@@ -273,6 +274,7 @@ pub fn get_command(
273274
}
274275
}
275276
Some(("push", push_matches)) => Box::new(PushCommand::new(
277+
cli_catalog.get_one()?,
276278
cli_catalog.get_one()?,
277279
cli_catalog.get_one()?,
278280
push_matches
@@ -554,15 +556,15 @@ fn validate_dataset_ref_pattern(
554556
let valid_ref = validate_dataset_ref(catalog, dsr)?;
555557
Ok(DatasetRefPattern::Ref(valid_ref))
556558
}
557-
DatasetRefPattern::Pattern(an, drp) => {
559+
DatasetRefPattern::Pattern(drp) => {
558560
let workspace_svc = catalog.get_one::<WorkspaceService>()?;
559-
if !workspace_svc.is_multi_tenant_workspace() && an.is_some() {
561+
if !workspace_svc.is_multi_tenant_workspace() && drp.account_name.is_some() {
560562
return Err(MultiTenantRefUnexpectedError {
561-
dataset_ref_pattern: DatasetRefPattern::Pattern(an, drp),
563+
dataset_ref_pattern: DatasetRefPattern::Pattern(drp),
562564
}
563565
.into());
564566
}
565-
Ok(DatasetRefPattern::Pattern(an, drp))
567+
Ok(DatasetRefPattern::Pattern(drp))
566568
}
567569
}
568570
}

‎src/app/cli/src/cli_parser.rs‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -702,7 +702,7 @@ pub fn cli() -> Command {
702702
Arg::new("dataset")
703703
.action(ArgAction::Append)
704704
.index(1)
705-
.value_parser(value_parse_dataset_ref_any)
705+
.value_parser(value_parse_dataset_ref_pattern_any)
706706
.help("Local or remote dataset reference(s)"),
707707
Arg::new("as")
708708
.long("as")
@@ -737,6 +737,10 @@ pub fn cli() -> Command {
737737
738738
kamu pull org.example.data
739739
740+
Fetch latest data in datasets matching pattern:
741+
742+
kamu pull org.example.%
743+
740744
Fetch latest data for the entire dependency tree of a dataset:
741745
742746
kamu pull --recursive org.example.derivative
@@ -780,7 +784,7 @@ pub fn cli() -> Command {
780784
Arg::new("dataset")
781785
.action(ArgAction::Append)
782786
.index(1)
783-
.value_parser(value_parse_dataset_ref_any)
787+
.value_parser(value_parse_dataset_ref_pattern_any)
784788
.help("Local or remote dataset reference(s)"),
785789
Arg::new("to")
786790
.long("to")
@@ -813,6 +817,10 @@ pub fn cli() -> Command {
813817
814818
kamu push org.example.data
815819
820+
Sync datasets matching pattern that already have push aliases:
821+
822+
kamu push org.example.%
823+
816824
Add dataset to local IPFS node and update IPNS entry to the new CID:
817825
818826
kamu push org.example.data --to ipns://k5..zy

‎src/app/cli/src/cli_value_parser.rs‎

Lines changed: 13 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ use std::str::FromStr;
1212
use opendatafabric::{
1313
DatasetName,
1414
DatasetRef,
15-
DatasetRefAny,
15+
DatasetRefAnyPattern,
1616
DatasetRefPattern,
1717
DatasetRefRemote,
1818
Multihash,
@@ -22,7 +22,7 @@ use url::Url;
2222

2323
pub(crate) fn value_parse_dataset_ref_pattern_local(s: &str) -> Result<DatasetRefPattern, String> {
2424
match DatasetRefPattern::from_str(s) {
25-
Ok(drp) => Ok(drp),
25+
Ok(dataset_ref_pattern) => Ok(dataset_ref_pattern),
2626
Err(_) => Err(
2727
"Local reference should be in form: `did:odf:...`, `my.dataset.id`, or a wildcard \
2828
pattern `my.dataset.%`"
@@ -31,6 +31,17 @@ pub(crate) fn value_parse_dataset_ref_pattern_local(s: &str) -> Result<DatasetRe
3131
}
3232
}
3333

34+
pub(crate) fn value_parse_dataset_ref_pattern_any(s: &str) -> Result<DatasetRefAnyPattern, String> {
35+
match DatasetRefAnyPattern::from_str(s) {
36+
Ok(dataset_ref_pattern) => Ok(dataset_ref_pattern),
37+
Err(_) => Err("Dataset reference should be in form: `my.dataset.id` or \
38+
`repository/account/dataset-id` or `did:odf:...` or `scheme://some-url` \
39+
or a wildcard pattern: `my.dataset.%` or \
40+
`repository/account/remote.dataset.%`"
41+
.to_string()),
42+
}
43+
}
44+
3445
pub(crate) fn value_parse_dataset_name(s: &str) -> Result<DatasetName, String> {
3546
match DatasetName::try_from(s) {
3647
Ok(v) => Ok(v),
@@ -59,15 +70,6 @@ pub(crate) fn value_parse_dataset_ref_remote(s: &str) -> Result<DatasetRefRemote
5970
}
6071
}
6172

62-
pub(crate) fn value_parse_dataset_ref_any(s: &str) -> Result<DatasetRefAny, String> {
63-
match DatasetRefAny::try_from(s) {
64-
Ok(v) => Ok(v),
65-
Err(_) => Err("Dataset reference should be in form: `my.dataset.id` or \
66-
`repository/account/dataset-id` or `did:odf:...` or `scheme://some-url`"
67-
.to_string()),
68-
}
69-
}
70-
7173
pub(crate) fn value_parse_repo_name(s: &str) -> Result<RepoName, String> {
7274
match RepoName::try_from(s) {
7375
Ok(v) => Ok(v),

‎src/app/cli/src/commands/delete_command.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ use std::sync::Arc;
1111

1212
use futures::{future, StreamExt, TryStreamExt};
1313
use kamu::domain::*;
14-
use kamu::utils::datasets_filtering::filter_datasets_by_pattern;
14+
use kamu::utils::datasets_filtering::filter_datasets_by_local_pattern;
1515
use opendatafabric::*;
1616

1717
use super::{common, CLIError, Command};
@@ -58,7 +58,7 @@ impl Command for DeleteCommand {
5858
let dataset_refs: Vec<_> = if self.all {
5959
unimplemented!("All deletion is not yet supported")
6060
} else {
61-
let dataset_ids: Vec<_> = filter_datasets_by_pattern(
61+
let dataset_ids: Vec<_> = filter_datasets_by_local_pattern(
6262
self.dataset_repo.as_ref(),
6363
self.dataset_ref_patterns.clone(),
6464
)

‎src/app/cli/src/commands/pull_command.rs‎

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,9 @@ use std::sync::{Arc, Mutex};
1111
use std::time::Duration;
1212

1313
use chrono::{DateTime, Utc};
14+
use futures::TryStreamExt;
1415
use kamu::domain::*;
16+
use kamu::utils::datasets_filtering::filter_datasets_by_any_pattern;
1517
use opendatafabric::*;
1618

1719
use super::{BatchError, CLIError, Command};
@@ -24,8 +26,9 @@ use crate::output::OutputConfig;
2426
pub struct PullCommand {
2527
pull_svc: Arc<dyn PullService>,
2628
dataset_repo: Arc<dyn DatasetRepository>,
29+
search_svc: Arc<dyn SearchService>,
2730
output_config: Arc<OutputConfig>,
28-
refs: Vec<DatasetRefAny>,
31+
refs: Vec<DatasetRefAnyPattern>,
2932
all: bool,
3033
recursive: bool,
3134
fetch_uncacheable: bool,
@@ -38,6 +41,7 @@ impl PullCommand {
3841
pub fn new<I>(
3942
pull_svc: Arc<dyn PullService>,
4043
dataset_repo: Arc<dyn DatasetRepository>,
44+
search_svc: Arc<dyn SearchService>,
4145
output_config: Arc<OutputConfig>,
4246
refs: I,
4347
all: bool,
@@ -48,11 +52,12 @@ impl PullCommand {
4852
force: bool,
4953
) -> Self
5054
where
51-
I: IntoIterator<Item = DatasetRefAny>,
55+
I: IntoIterator<Item = DatasetRefAnyPattern>,
5256
{
5357
Self {
5458
pull_svc,
5559
dataset_repo,
60+
search_svc,
5661
output_config,
5762
refs: refs.into_iter().collect(),
5863
all,
@@ -69,9 +74,16 @@ impl PullCommand {
6974
listener: Option<Arc<dyn PullMultiListener>>,
7075
) -> Result<Vec<PullResponse>, CLIError> {
7176
let local_name = self.as_name.as_ref().unwrap();
72-
let remote_ref = self.refs[0].as_remote_ref(|_| true).map_err(|_| {
73-
CLIError::usage_error("When using --as reference should point to a remote dataset")
74-
})?;
77+
let remote_ref = match self.refs[0].as_dataset_ref_any() {
78+
Some(dataset_ref_any) => dataset_ref_any.as_remote_ref(|_| true).map_err(|_| {
79+
CLIError::usage_error("When using --as reference should point to a remote dataset")
80+
})?,
81+
None => {
82+
return Err(CLIError::usage_error(
83+
"When using --as reference should not point to wildcard pattern",
84+
))
85+
}
86+
};
7587

7688
Ok(self
7789
.pull_svc
@@ -93,10 +105,18 @@ impl PullCommand {
93105
&self,
94106
listener: Option<Arc<dyn PullMultiListener>>,
95107
) -> Result<Vec<PullResponse>, CLIError> {
108+
let dataset_refs: Vec<_> = filter_datasets_by_any_pattern(
109+
self.dataset_repo.as_ref(),
110+
self.search_svc.clone(),
111+
self.refs.clone(),
112+
)
113+
.try_collect()
114+
.await?;
115+
96116
Ok(self
97117
.pull_svc
98118
.pull_multi(
99-
self.refs.clone(),
119+
dataset_refs,
100120
PullMultiOptions {
101121
recursive: self.recursive,
102122
all: self.all,

‎src/app/cli/src/commands/push_command.rs‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,9 @@
1010
use std::sync::{Arc, Mutex};
1111
use std::time::Duration;
1212

13+
use futures::TryStreamExt;
1314
use kamu::domain::*;
15+
use kamu::utils::datasets_filtering::filter_datasets_by_any_pattern;
1416
use opendatafabric::*;
1517

1618
use super::{BatchError, CLIError, Command};
@@ -23,7 +25,8 @@ use crate::output::OutputConfig;
2325
pub struct PushCommand {
2426
push_svc: Arc<dyn PushService>,
2527
dataset_repo: Arc<dyn DatasetRepository>,
26-
refs: Vec<DatasetRefAny>,
28+
search_svc: Arc<dyn SearchService>,
29+
refs: Vec<DatasetRefAnyPattern>,
2730
all: bool,
2831
recursive: bool,
2932
add_aliases: bool,
@@ -36,6 +39,7 @@ impl PushCommand {
3639
pub fn new<I>(
3740
push_svc: Arc<dyn PushService>,
3841
dataset_repo: Arc<dyn DatasetRepository>,
42+
search_svc: Arc<dyn SearchService>,
3943
refs: I,
4044
all: bool,
4145
recursive: bool,
@@ -45,11 +49,12 @@ impl PushCommand {
4549
output_config: Arc<OutputConfig>,
4650
) -> Self
4751
where
48-
I: Iterator<Item = DatasetRefAny>,
52+
I: Iterator<Item = DatasetRefAnyPattern>,
4953
{
5054
Self {
5155
push_svc,
5256
dataset_repo,
57+
search_svc,
5358
refs: refs.collect(),
5459
all,
5560
recursive,
@@ -65,13 +70,20 @@ impl PushCommand {
6570
listener: Option<Arc<dyn SyncMultiListener>>,
6671
) -> Result<Vec<PushResponse>, CLIError> {
6772
if let Some(remote_ref) = &self.to {
68-
let local_ref = self.refs[0]
69-
.as_local_ref(|_| !self.dataset_repo.is_multi_tenant())
70-
.map_err(|_| {
71-
CLIError::usage_error(
72-
"When using --to reference should point to a local dataset",
73-
)
74-
})?;
73+
let local_ref = match self.refs[0].as_dataset_ref_any() {
74+
Some(dataset_ref_any) => dataset_ref_any
75+
.as_local_ref(|_| !self.dataset_repo.is_multi_tenant())
76+
.map_err(|_| {
77+
CLIError::usage_error(
78+
"When using --to reference should point to a local dataset",
79+
)
80+
})?,
81+
None => {
82+
return Err(CLIError::usage_error(
83+
"When using --to reference should not point to wildcard pattern",
84+
))
85+
}
86+
};
7587

7688
Ok(self
7789
.push_svc
@@ -90,10 +102,18 @@ impl PushCommand {
90102
)
91103
.await)
92104
} else {
105+
let dataset_refs: Vec<_> = filter_datasets_by_any_pattern(
106+
self.dataset_repo.as_ref(),
107+
self.search_svc.clone(),
108+
self.refs.clone(),
109+
)
110+
.try_collect()
111+
.await?;
112+
93113
Ok(self
94114
.push_svc
95115
.push_multi(
96-
self.refs.clone(),
116+
dataset_refs,
97117
PushMultiOptions {
98118
all: self.all,
99119
recursive: self.recursive,

‎src/app/cli/src/commands/set_watermark_command.rs‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ pub struct SetWatermarkCommand {
1919
dataset_repo: Arc<dyn DatasetRepository>,
2020
remote_alias_reg: Arc<dyn RemoteAliasesRegistry>,
2121
pull_svc: Arc<dyn PullService>,
22-
refs: Vec<DatasetRefAny>,
22+
refs: Vec<DatasetRefAnyPattern>,
2323
all: bool,
2424
recursive: bool,
2525
watermark: String,
@@ -37,7 +37,7 @@ impl SetWatermarkCommand {
3737
) -> Self
3838
where
3939
S: Into<String>,
40-
I: Iterator<Item = DatasetRefAny>,
40+
I: Iterator<Item = DatasetRefAnyPattern>,
4141
{
4242
Self {
4343
dataset_repo,
@@ -58,11 +58,17 @@ impl Command for SetWatermarkCommand {
5858
return Err(CLIError::usage_error(
5959
"Only one dataset can be provided when setting a watermark",
6060
));
61-
} else if self.recursive || self.all {
61+
}
62+
if self.recursive || self.all {
6263
return Err(CLIError::usage_error(
6364
"Can't use --all or --recursive flags when setting a watermark",
6465
));
6566
}
67+
if self.refs[0].is_pattern() {
68+
return Err(CLIError::usage_error(
69+
"Cannot use a pattern when setting a watermark",
70+
));
71+
}
6672

6773
let watermark = DateTime::parse_from_rfc3339(&self.watermark).map_err(|_| {
6874
CLIError::usage_error(format!(
@@ -72,6 +78,8 @@ impl Command for SetWatermarkCommand {
7278
})?;
7379

7480
let dataset_ref = self.refs[0]
81+
.as_dataset_ref_any()
82+
.unwrap()
7583
.as_local_ref(|_| !self.dataset_repo.is_multi_tenant())
7684
.map_err(|_| CLIError::usage_error("Expected a local dataset reference"))?;
7785

0 commit comments

Comments
 (0)