Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions docs/docs/sources/azureblob.md
Original file line number Diff line number Diff line change
Expand Up @@ -63,15 +63,19 @@ The spec takes the following fields:
* `excluded_patterns` (`list[str]`, optional): a list of glob patterns to exclude files, e.g. `["*.tmp", "**/*.log"]`.
Any file or directory matching these patterns will be excluded even if they match `included_patterns`.
If not specified, no files will be excluded.
* `sas_token` (`cocoindex.TransientAuthEntryReference[str]`, optional): a SAS token for authentication.
* `account_access_key` (`cocoindex.TransientAuthEntryReference[str]`, optional): an account access key for authentication.

:::info

`included_patterns` and `excluded_patterns` are using Unix-style glob syntax. See [globset syntax](https://docs.rs/globset/latest/globset/index.html#syntax) for the details.

:::

* `max_file_size` (`int`, optional): if provided, files exceeding this size in bytes will be treated as non-existent and skipped during processing.
This is useful to avoid processing large files that are not relevant to your use case, such as videos or backups.
If not specified, no size limit is applied.
* `sas_token` (`cocoindex.TransientAuthEntryReference[str]`, optional): a SAS token for authentication.
* `account_access_key` (`cocoindex.TransientAuthEntryReference[str]`, optional): an account access key for authentication.

### Schema

The output is a [*KTable*](/docs/core/data_types#ktable) with the following sub fields:
Expand Down
1 change: 1 addition & 0 deletions python/cocoindex/sources/_engine_builtin_specs.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ class AzureBlob(op.SourceSpec):
binary: bool = False
included_patterns: list[str] | None = None
excluded_patterns: list[str] | None = None
max_file_size: int | None = None

sas_token: TransientAuthEntryReference[str] | None = None
account_access_key: TransientAuthEntryReference[str] | None = None
Expand Down
26 changes: 26 additions & 0 deletions src/ops/sources/azure_blob.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ pub struct Spec {
binary: bool,
included_patterns: Option<Vec<String>>,
excluded_patterns: Option<Vec<String>>,
max_file_size: Option<i64>,

/// SAS token for authentication. Takes precedence over account_access_key.
sas_token: Option<AuthEntryReference<String>>,
Expand All @@ -32,6 +33,7 @@ struct Executor {
prefix: Option<String>,
binary: bool,
pattern_matcher: PatternMatcher,
max_file_size: Option<i64>,
}

fn datetime_to_ordinal(dt: &time::OffsetDateTime) -> Ordinal {
Expand Down Expand Up @@ -73,6 +75,13 @@ impl SourceExecutor for Executor {
// Only include files (not directories)
if key.ends_with('/') { continue; }

// Check file size limit
if let Some(max_size) = self.max_file_size {
if blob.properties.content_length > max_size as u64 {
continue;
}
}

if self.pattern_matcher.is_file_included(key) {
let ordinal = Some(datetime_to_ordinal(&blob.properties.last_modified));
batch.push(PartialSourceRow {
Expand Down Expand Up @@ -115,6 +124,22 @@ impl SourceExecutor for Executor {
});
}

// Check file size limit
if let Some(max_size) = self.max_file_size {
let blob_client = self
.client
.container_client(&self.container_name)
.blob_client(key_str.as_ref());
let properties = blob_client.get_properties().await?;
if properties.blob.properties.content_length > max_size as u64 {
return Ok(PartialSourceRowData {
value: Some(SourceValue::NonExistence),
ordinal: Some(Ordinal::unavailable()),
content_version_fp: None,
});
}
}

let blob_client = self
.client
.container_client(&self.container_name)
Expand Down Expand Up @@ -238,6 +263,7 @@ impl SourceFactoryBase for Factory {
prefix: spec.prefix,
binary: spec.binary,
pattern_matcher: PatternMatcher::new(spec.included_patterns, spec.excluded_patterns)?,
max_file_size: spec.max_file_size,
}))
}
}
Loading