1use std::path::PathBuf;
19use std::sync::Arc;
20
21use log::debug;
22
23use super::HF_SCHEME;
24use super::config::HfConfig;
25use super::core::HfCore;
26use super::core::HfDownloadMode;
27use super::core::{HfRepo, HfRepoType};
28use super::deleter::HfDeleter;
29use super::lister::HfLister;
30use super::reader::*;
31use super::writer::HfLazyWriter;
32use opendal_core::raw::*;
33use opendal_core::*;
34
35#[doc = include_str!("docs.md")]
37#[derive(Debug, Default)]
38pub struct HfBuilder {
39 pub(super) config: HfConfig,
40}
41
42impl HfBuilder {
43 pub fn repo_type(mut self, repo_type: &str) -> Self {
54 if !repo_type.is_empty()
55 && let Ok(rt) = HfRepoType::parse(repo_type)
56 {
57 self.config.repo_type = Some(rt);
58 }
59 self
60 }
61
62 pub fn repo_id(mut self, repo_id: &str) -> Self {
72 if !repo_id.is_empty() {
73 self.config.repo_id = Some(repo_id.to_string());
74 }
75 self
76 }
77
78 pub fn revision(mut self, revision: &str) -> Self {
86 if !revision.is_empty() {
87 self.config.revision = Some(revision.to_string());
88 }
89 self
90 }
91
92 pub fn root(mut self, root: &str) -> Self {
96 self.config.root = if root.is_empty() {
97 None
98 } else {
99 Some(root.to_string())
100 };
101
102 self
103 }
104
105 pub fn token(mut self, token: &str) -> Self {
109 if !token.is_empty() {
110 self.config.token = Some(token.to_string());
111 }
112 self
113 }
114
115 pub fn download_mode(mut self, mode: &str) -> Self {
128 if !mode.is_empty()
129 && let Ok(m) = HfDownloadMode::parse(mode)
130 {
131 self.config.download_mode = Some(m);
132 }
133 self
134 }
135
136 pub fn endpoint(mut self, endpoint: &str) -> Self {
143 if !endpoint.is_empty() {
144 self.config.endpoint = Some(endpoint.to_string());
145 }
146 self
147 }
148
149 fn hf_endpoint(&self) -> String {
150 self.config
151 .endpoint
152 .clone()
153 .or_else(|| std::env::var("HF_ENDPOINT").ok())
154 .unwrap_or_else(|| "https://huggingface.co".to_string())
155 }
156
157 fn hf_download_mode(&self) -> HfDownloadMode {
160 if let Some(mode) = self.config.download_mode {
161 return mode;
162 }
163 if let Ok(val) = std::env::var("HF_HUB_DISABLE_XET")
164 && !val.is_empty()
165 {
166 return HfDownloadMode::Http;
167 }
168 HfDownloadMode::default()
169 }
170
171 fn hf_home() -> Option<PathBuf> {
172 if let Ok(h) = std::env::var("HF_HOME") {
173 return Some(PathBuf::from(h));
174 }
175 if let Ok(xdg) = std::env::var("XDG_CACHE_HOME") {
176 return Some(PathBuf::from(xdg).join("huggingface"));
177 }
178 let home = std::env::var("HOME").ok()?;
179 Some(PathBuf::from(home).join(".cache/huggingface"))
180 }
181
182 fn hf_token(&self) -> Option<String> {
185 if let Some(t) = self.config.token.clone() {
186 return Some(t);
187 }
188 if let Ok(val) = std::env::var("HF_HUB_DISABLE_IMPLICIT_TOKEN")
189 && !val.is_empty()
190 {
191 return None;
192 }
193 if let Ok(t) = std::env::var("HF_TOKEN")
194 && !t.is_empty()
195 {
196 return Some(t);
197 }
198 let token_path = if let Ok(p) = std::env::var("HF_TOKEN_PATH") {
199 Some(PathBuf::from(p))
200 } else {
201 Self::hf_home().map(|h| h.join("token"))
202 };
203 token_path
204 .and_then(|p| std::fs::read_to_string(p).ok())
205 .map(|s| s.trim().to_string())
206 .filter(|s| !s.is_empty())
207 }
208}
209
210impl Builder for HfBuilder {
211 type Config = HfConfig;
212
213 fn build(self) -> Result<impl Service> {
214 debug!("backend build started: {:?}", self);
215
216 let token = self.hf_token();
217 let endpoint = self.hf_endpoint();
218 let download_mode = self.hf_download_mode();
219
220 let repo_type = self.config.repo_type.ok_or_else(|| {
221 Error::new(ErrorKind::ConfigInvalid, "repo_type is required")
222 .with_operation("Builder::build")
223 .with_context("service", HF_SCHEME)
224 })?;
225 debug!("backend use repo_type: {:?}", repo_type);
226
227 let repo_id = self.config.repo_id.ok_or_else(|| {
228 Error::new(ErrorKind::ConfigInvalid, "repo_id is required")
229 .with_operation("Builder::build")
230 .with_context("service", HF_SCHEME)
231 })?;
232 debug!("backend use repo_id: {}", repo_id);
233
234 let revision = match &self.config.revision {
235 Some(revision) => revision.clone(),
236 None => "main".to_string(),
237 };
238 debug!("backend use revision: {}", revision);
239
240 let root = normalize_root(&self.config.root.unwrap_or_default());
241 debug!("backend use root: {}", root);
242 debug!("backend use token: {}", token.is_some());
243 debug!("backend use endpoint: {}", endpoint);
244 debug!("backend use download_mode: {:?}", download_mode);
245
246 let info = ServiceInfo::new(HF_SCHEME, "", "");
247 let capability = Capability {
248 stat: true,
249 read: true,
250 write: token.is_some(),
251 write_can_multi: token.is_some(),
252 delete: token.is_some(),
253 delete_max_size: Some(100),
254 list: true,
255 list_with_recursive: true,
256 shared: true,
257 ..Default::default()
258 };
259
260 let repo = HfRepo::new(repo_type, repo_id, Some(revision.clone()));
261 debug!("backend repo uri: {:?}", repo.uri(&root, ""));
262
263 Ok(HfBackend {
264 core: Arc::new(HfCore::build(
265 info,
266 capability,
267 repo,
268 root,
269 token,
270 endpoint,
271 download_mode,
272 )?),
273 })
274 }
275}
276
277#[derive(Debug, Clone)]
279pub struct HfBackend {
280 pub(crate) core: Arc<HfCore>,
281}
282
283impl Service for HfBackend {
284 type Reader = oio::StreamReader<HfReader>;
285 type Writer = HfLazyWriter;
286 type Lister = oio::PageLister<HfLister>;
287 type Deleter = oio::BatchDeleter<HfDeleter>;
288 type Copier = ();
289
290 fn info(&self) -> ServiceInfo {
291 self.core.info.clone()
292 }
293
294 fn capability(&self) -> Capability {
295 self.core.capability
296 }
297
298 async fn create_dir(
299 &self,
300 _ctx: &OperationContext,
301 _path: &str,
302 _args: OpCreateDir,
303 ) -> Result<RpCreateDir> {
304 Err(Error::new(
305 ErrorKind::Unsupported,
306 "operation is not supported",
307 ))
308 }
309
310 async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
311 if path == "/" {
313 return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
314 }
315
316 if self.core.repo.is_bucket() && path.ends_with('/') {
318 return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
319 }
320
321 let info = self.core.path_info(ctx, path).await?;
322 Ok(RpStat::new(info.metadata()?))
323 }
324 fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
325 let output: oio::StreamReader<HfReader> = {
326 Ok(oio::StreamReader::new(HfReader::new(
327 self.clone(),
328 ctx.clone(),
329 path,
330 args,
331 )))
332 }?;
333
334 Ok(output)
335 }
336
337 fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
338 let output: oio::PageLister<HfLister> = {
339 let lister = HfLister::new(
340 self.core.clone(),
341 ctx.clone(),
342 path.to_string(),
343 args.recursive(),
344 );
345 Ok(oio::PageLister::new(lister))
346 }?;
347
348 Ok(output)
349 }
350
351 fn write(&self, ctx: &OperationContext, path: &str, _args: OpWrite) -> Result<Self::Writer> {
352 Ok(HfLazyWriter::new(
353 self.core.clone(),
354 ctx.clone(),
355 path.to_string(),
356 ))
357 }
358
359 fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
360 let output: oio::BatchDeleter<HfDeleter> = {
361 let deleter = HfDeleter::new(self.core.clone(), ctx.clone());
362 let max_batch_size = self.core.capability.delete_max_size;
363 Ok(oio::BatchDeleter::new(deleter, max_batch_size))
364 }?;
365
366 Ok(output)
367 }
368
369 fn copy(
370 &self,
371 _ctx: &OperationContext,
372 _from: &str,
373 _to: &str,
374 _args: OpCopy,
375 _opts: OpCopier,
376 ) -> Result<Self::Copier> {
377 Err(Error::new(
378 ErrorKind::Unsupported,
379 "operation is not supported",
380 ))
381 }
382
383 async fn rename(
384 &self,
385 _ctx: &OperationContext,
386 _from: &str,
387 _to: &str,
388 _args: OpRename,
389 ) -> Result<RpRename> {
390 Err(Error::new(
391 ErrorKind::Unsupported,
392 "operation is not supported",
393 ))
394 }
395
396 async fn presign(
397 &self,
398 _ctx: &OperationContext,
399 _path: &str,
400 _args: OpPresign,
401 ) -> Result<RpPresign> {
402 Err(Error::new(
403 ErrorKind::Unsupported,
404 "operation is not supported",
405 ))
406 }
407}
408
409#[cfg(test)]
410pub(super) mod test_utils {
411 use std::sync::Arc;
412
413 use super::super::core::{HfCore, HfDownloadMode};
414 use super::super::core::{HfRepo, HfRepoType};
415 use super::HfBuilder;
416 use opendal_core::Capability;
417 use opendal_core::HttpTransporter;
418 use opendal_core::OperationContext;
419 use opendal_core::Operator;
420 use opendal_core::raw::ServiceInfo;
421
422 fn finish_operator(op: Operator) -> Operator {
423 let transport =
424 HttpTransporter::new(opendal_http_transport_reqwest::ReqwestTransport::default());
425 op.with_context(OperationContext::new().with_http_transport(transport))
426 }
427
428 pub fn mbpp_operator() -> Operator {
429 let op = Operator::new(
430 HfBuilder::default()
431 .repo_type("dataset")
432 .repo_id("google-research-datasets/mbpp"),
433 )
434 .unwrap();
435 finish_operator(op)
436 }
437
438 pub fn testing_dataset_core() -> Arc<HfCore> {
439 let repo_id = std::env::var("HF_OPENDAL_DATASET").expect("HF_OPENDAL_DATASET must be set");
440 let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN must be set");
441
442 let info = ServiceInfo::new("hf", "", "");
443 let capability = Capability {
444 read: true,
445 write: true,
446 delete: true,
447 ..Default::default()
448 };
449
450 let repo = HfRepo::new(HfRepoType::Dataset, repo_id, Some("main".to_string()));
451
452 Arc::new(
453 HfCore::build(
454 info,
455 capability,
456 repo,
457 "/".to_string(),
458 Some(token),
459 "https://huggingface.co".to_string(),
460 HfDownloadMode::Xet,
461 )
462 .expect("failed to build HfCore"),
463 )
464 }
465
466 pub fn testing_bucket_operator() -> Operator {
467 let repo_id = std::env::var("HF_OPENDAL_BUCKET").expect("HF_OPENDAL_BUCKET must be set");
468 let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN must be set");
469 let op = Operator::new(
470 HfBuilder::default()
471 .repo_type("bucket")
472 .repo_id(&repo_id)
473 .token(&token),
474 )
475 .unwrap();
476 finish_operator(op)
477 }
478}
479
480#[cfg(test)]
481mod tests {
482 use super::*;
483 use std::sync::Mutex;
484
485 static ENV_LOCK: Mutex<()> = Mutex::new(());
487
488 fn builder_with_token(token: &str) -> HfBuilder {
489 HfBuilder::default().token(token)
490 }
491
492 fn builder_no_token() -> HfBuilder {
493 HfBuilder::default()
494 }
495
496 #[test]
497 fn hf_token_config_takes_priority_over_env() {
498 let _guard = ENV_LOCK.lock().unwrap();
499 unsafe { std::env::set_var("HF_TOKEN", "env-token") };
500 let result = builder_with_token("config-token").hf_token();
501 unsafe { std::env::remove_var("HF_TOKEN") };
502 assert_eq!(result.as_deref(), Some("config-token"));
503 }
504
505 #[test]
506 fn hf_token_reads_hf_token_env_var() {
507 let _guard = ENV_LOCK.lock().unwrap();
508 unsafe { std::env::remove_var("HF_HUB_DISABLE_IMPLICIT_TOKEN") };
509 unsafe { std::env::remove_var("HF_TOKEN_PATH") };
510 unsafe { std::env::set_var("HF_TOKEN", "my-env-token") };
511 let result = builder_no_token().hf_token();
512 unsafe { std::env::remove_var("HF_TOKEN") };
513 assert_eq!(result.as_deref(), Some("my-env-token"));
514 }
515
516 #[test]
517 fn hf_token_disable_flag_suppresses_discovery() {
518 let _guard = ENV_LOCK.lock().unwrap();
519 unsafe { std::env::set_var("HF_HUB_DISABLE_IMPLICIT_TOKEN", "1") };
520 unsafe { std::env::set_var("HF_TOKEN", "my-env-token") };
521 let result = builder_no_token().hf_token();
522 unsafe { std::env::remove_var("HF_HUB_DISABLE_IMPLICIT_TOKEN") };
523 unsafe { std::env::remove_var("HF_TOKEN") };
524 assert_eq!(result, None);
525 }
526
527 #[test]
528 fn hf_token_reads_from_file_via_hf_token_path() {
529 let _guard = ENV_LOCK.lock().unwrap();
530 let token_file = std::env::temp_dir().join("opendal-hf-token-test");
531 std::fs::write(&token_file, "file-token\n").unwrap();
532 unsafe { std::env::remove_var("HF_HUB_DISABLE_IMPLICIT_TOKEN") };
533 unsafe { std::env::remove_var("HF_TOKEN") };
534 unsafe { std::env::set_var("HF_TOKEN_PATH", &token_file) };
535 let result = builder_no_token().hf_token();
536 unsafe { std::env::remove_var("HF_TOKEN_PATH") };
537 std::fs::remove_file(&token_file).ok();
538 assert_eq!(result.as_deref(), Some("file-token"));
539 }
540
541 #[test]
542 fn hf_endpoint_returns_default() {
543 let _guard = ENV_LOCK.lock().unwrap();
544 unsafe { std::env::remove_var("HF_ENDPOINT") };
545 let result = HfBuilder::default().hf_endpoint();
546 assert_eq!(result, "https://huggingface.co");
547 }
548
549 #[test]
550 fn hf_endpoint_config_takes_priority_over_env() {
551 let _guard = ENV_LOCK.lock().unwrap();
552 unsafe { std::env::set_var("HF_ENDPOINT", "https://env.example.com") };
553 let result = HfBuilder::default()
554 .endpoint("https://config.example.com")
555 .hf_endpoint();
556 unsafe { std::env::remove_var("HF_ENDPOINT") };
557 assert_eq!(result, "https://config.example.com");
558 }
559
560 #[test]
561 fn hf_endpoint_reads_hf_endpoint_env_var() {
562 let _guard = ENV_LOCK.lock().unwrap();
563 unsafe { std::env::set_var("HF_ENDPOINT", "https://env.example.com") };
564 let result = HfBuilder::default().hf_endpoint();
565 unsafe { std::env::remove_var("HF_ENDPOINT") };
566 assert_eq!(result, "https://env.example.com");
567 }
568
569 #[test]
570 fn hf_download_mode_defaults_to_xet() {
571 let _guard = ENV_LOCK.lock().unwrap();
572 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
573 assert_eq!(HfBuilder::default().hf_download_mode(), HfDownloadMode::Xet);
574 }
575
576 #[test]
577 fn hf_download_mode_disable_xet_env_forces_http() {
578 let _guard = ENV_LOCK.lock().unwrap();
579 unsafe { std::env::set_var("HF_HUB_DISABLE_XET", "1") };
580 let mode = HfBuilder::default().hf_download_mode();
581 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
582 assert_eq!(mode, HfDownloadMode::Http);
583 }
584
585 #[test]
586 fn hf_download_mode_config_takes_priority_over_env() {
587 let _guard = ENV_LOCK.lock().unwrap();
588 unsafe { std::env::set_var("HF_HUB_DISABLE_XET", "1") };
589 let mode = HfBuilder::default().download_mode("xet").hf_download_mode();
590 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
591 assert_eq!(mode, HfDownloadMode::Xet);
592 }
593
594 #[test]
595 fn hf_download_mode_empty_env_keeps_xet() {
596 let _guard = ENV_LOCK.lock().unwrap();
597 unsafe { std::env::set_var("HF_HUB_DISABLE_XET", "") };
598 let mode = HfBuilder::default().hf_download_mode();
599 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
600 assert_eq!(mode, HfDownloadMode::Xet);
601 }
602
603 #[test]
604 fn build_accepts_datasets_alias() {
605 HfBuilder::default()
606 .repo_id("org/repo")
607 .repo_type("datasets")
608 .build()
609 .expect("builder should accept datasets alias");
610 }
611
612 #[test]
613 fn build_accepts_space_repo_type() {
614 HfBuilder::default()
615 .repo_id("org/space")
616 .repo_type("space")
617 .build()
618 .expect("builder should accept space repo type");
619 }
620
621 #[test]
622 fn test_both_schemes_are_supported() {
623 use opendal_core::OperatorRegistry;
624
625 let registry = OperatorRegistry::get();
626 super::super::register_hf_service(registry);
627
628 let op = registry
630 .load("hf://user/repo")
631 .expect("short scheme should be registered and work");
632 assert_eq!(op.info().scheme(), "hf");
633
634 let op = registry
636 .load("huggingface://user/repo")
637 .expect("long scheme should be registered and work");
638 assert_eq!(op.info().scheme(), "hf");
639 }
640}