Support for backblaze B2 and minio

[ericmjl]

Hi everyone!

I was wondering what it might take to support backblaze B2 and minio? Both have s3-compatible APIs. If it is just a matter of changing environment variables and pointing to a different bucket URI, I am happy to contribute documentation on how to accomplish this. Asking this mostly b/c I'd like a way to take advantage of B2's zero egress fee storage, which AWS does not provide.

Cheers,
Eric

[isabelizimm]

Hello there! Thank you so much for this report, this is an interesting question 👀 Since pins is powered by fsspec, I did a bit of a peek into seeing what their minio and backblaze support looks like. It looks like there are a few different ways to authenticate for s3 compatible APIs. There is an option to use FSSPEC_S3_ENDPOINT_URL to set a minio URL that may work out of the box (see this issue for minio url examples). Are you able to try out this environment variable and see if it will authenticate?

If not, pins might need to accept storage_options on board creation for authentication, which seems doable!

[ericmjl]

Hi @isabelizimm! Thanks for responding, really appreciate it 😄.

I have tried the following within a notebook, which I think constitutes a minimal reproducible example, but that it appears that I don't have credentials set correctly. I will illustrate below.

To start, these are the contents of a .env file, with sensitive information redacted:

# contents of .env
FSSPEC_S3_KEY="0***********************3"
FSSPEC_S3_SECRET="K*****************************o"
FSSPEC_S3_ENDPOINT_URL="https://s3.**********.backblazeb2.com"

This is the contents of my first code cell in a notebook:

import pins
import os
from dotenv import load_dotenv
from pins.data import mtcars

load_dotenv(".env") # .env lives in the same directory as the notebook

board = pins.board_s3("pins-test")

mtcars.to_csv("/tmp/mtcars2.csv")
board.pin_upload("/tmp/mtcars2.csv", "mtcars")

And this is the error that I get:

Details

{
	"name": "NoCredentialsError",
	"message": "Unable to locate credentials",
	"stack": "---------------------------------------------------------------------------
NoCredentialsError                        Traceback (most recent call last)
Cell In[1], line 14
     11 board = pins.board_s3(\"pins-test\")
     13 mtcars.to_csv(\"/tmp/mtcars2.csv\")
---> 14 board.pin_upload(\"/tmp/mtcars2.csv\", \"mtcars\")
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/pins/boards.py:434, in BaseBoard.pin_upload(self, paths, name, title, description, metadata)

403 def pin_upload(

404     self,

405     paths: "str | list[str]",

(...)

409     metadata=None,

410 ):

411     """Write a pin based on paths to one or more files.

412

413     This method simply uploads the files given, so they can be downloaded later

(...)

431         This gets stored on the Meta.user field.

432     """

--> 434     return self._pin_store(

435         paths,

436         name,

437         type="file",

438         title=title,

439         description=description,

440         metadata=metadata,

441     )
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/pins/boards.py:287, in BaseBoard._pin_store(self, x, name, type, title, description, metadata, versioned, created)

284 if not self.fs.exists(dst_pin_path):

285     # equivalent to mkdirp, want to fail quietly in case of race conditions

286     try:

--> 287         self.fs.mkdir(dst_pin_path)

288     except FileExistsError:

289         pass
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/fsspec/asyn.py:118, in sync_wrapper..wrapper(*args, **kwargs)

115 @functools.wraps(func)

116 def wrapper(*args, **kwargs):

117     self = obj or args[0]

--> 118     return sync(self.loop, func, *args, **kwargs)
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/fsspec/asyn.py:103, in sync(loop, func, timeout, *args, **kwargs)

101     raise FSTimeoutError from return_result

102 elif isinstance(return_result, BaseException):

--> 103     raise return_result

104 else:

105     return return_result
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/fsspec/asyn.py:56, in _runner(event, coro, result, timeout)

54     coro = asyncio.wait_for(coro, timeout=timeout)

55 try:

---> 56     result[0] = await coro

57 except Exception as ex:

58     result[0] = ex
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/s3fs/core.py:922, in S3FileSystem._mkdir(self, path, acl, create_parents, **kwargs)

918 if region_name:

919     params["CreateBucketConfiguration"] = {

920         "LocationConstraint": region_name

921     }

--> 922 await self._call_s3("create_bucket", **params)

923 self.invalidate_cache("")

924 self.invalidate_cache(bucket)
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/s3fs/core.py:365, in S3FileSystem._call_s3(self, method, *akwarglist, **kwargs)

363 logger.debug("CALL: %s - %s - %s", method.name, akwarglist, kw2)

364 additional_kwargs = self._get_s3_method_kwargs(method, *akwarglist, **kwargs)

--> 365 return await _error_wrapper(

366     method, kwargs=additional_kwargs, retries=self.retries

367 )
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/s3fs/core.py:145, in _error_wrapper(func, args, kwargs, retries)

143         err = e

144 err = translate_boto_error(err)

--> 145 raise err
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/s3fs/core.py:113, in _error_wrapper(func, args, kwargs, retries)

111 for i in range(retries):

112     try:

--> 113         return await func(*args, **kwargs)

114     except S3_RETRYABLE_ERRORS as e:

115         err = e
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/client.py:391, in AioBaseClient._make_api_call(self, operation_name, api_params)

387     maybe_compress_request(

388         self.meta.config, request_dict, operation_model

389     )

390     apply_request_checksum(request_dict)

--> 391     http, parsed_response = await self._make_request(

392         operation_model, request_dict, request_context

393     )

395 await self.meta.events.emit(

396     'after-call.{service_id}.{operation_name}'.format(

397         service_id=service_id, operation_name=operation_name

(...)

402     context=request_context,

403 )

405 if http.status_code >= 300:
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/client.py:419, in AioBaseClient._make_request(self, operation_model, request_dict, request_context)

415 async def _make_request(

416     self, operation_model, request_dict, request_context

417 ):

418     try:

--> 419         return await self._endpoint.make_request(

420             operation_model, request_dict

421         )

422     except Exception as e:

423         await self.meta.events.emit(

424             'after-call-error.{service_id}.{operation_name}'.format(

425                 service_id=self._service_model.service_id.hyphenize(),

(...)

429             context=request_context,

430         )
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/endpoint.py:98, in AioEndpoint._send_request(self, request_dict, operation_model)

96 context = request_dict['context']

97 self._update_retries_context(context, attempts)

---> 98 request = await self.create_request(request_dict, operation_model)

99 success_response, exception = await self._get_response(

100     request, operation_model, context

101 )

102 while await self._needs_retry(

103     attempts,

104     operation_model,

(...)

107     exception,

108 ):
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/endpoint.py:86, in AioEndpoint.create_request(self, params, operation_model)

82     service_id = operation_model.service_model.service_id.hyphenize()

83     event_name = 'request-created.{service_id}.{op_name}'.format(

84         service_id=service_id, op_name=operation_model.name

85     )

---> 86     await self._event_emitter.emit(

87         event_name,

88         request=request,

89         operation_name=operation_model.name,

90     )

91 prepared_request = self.prepare_request(request)

92 return prepared_request
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/hooks.py:66, in AioHierarchicalEmitter._emit(self, event_name, kwargs, stop_on_response)

63 logger.debug('Event %s: calling handler %s', event_name, handler)

65 # Await the handler if its a coroutine.

---> 66 response = await resolve_awaitable(handler(**kwargs))

67 responses.append((handler, response))

68 if stop_on_response and response is not None:
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/_helpers.py:15, in resolve_awaitable(obj)

13 async def resolve_awaitable(obj):

14     if inspect.isawaitable(obj):

---> 15         return await obj

17     return obj
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/signers.py:24, in AioRequestSigner.handler(self, operation_name, request, **kwargs)

19 async def handler(self, operation_name=None, request=None, **kwargs):

20     # This is typically hooked up to the "request-created" event

21     # from a client's event emitter.  When a new request is created

22     # this method is invoked to sign the request.

23     # Don't call this method directly.

---> 24     return await self.sign(operation_name, request)
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/aiobotocore/signers.py:92, in AioRequestSigner.sign(self, operation_name, request, region_name, signing_type, expires_in, signing_name)

89     else:

90         raise e

---> 92 auth.add_auth(request)
File ~/anaconda/envs/pins-test/lib/python3.11/site-packages/botocore/auth.py:418, in SigV4Auth.add_auth(self, request)

416 def add_auth(self, request):

417     if self.credentials is None:

--> 418         raise NoCredentialsError()

419     datetime_now = datetime.datetime.utcnow()

420     request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
NoCredentialsError: Unable to locate credentials"

}

There is a workaround, which is to follow backblaze's instructions for integrating s3fs with b2 -- essentially mounting the filesystem locally first and then using board_folder instead of board_s3, though I will admit this is a bit of a hack.

At this point, what do you think would be the best course of action? If there's something you think I could help chip in with, please let me know!

[machow]

This is super interesting! It seems like once the fsspec piece is locked down, wrapping it in a pins Board should be pretty quick. Are you able to list or download files directly using boto or fsspec?

It looks like this issue has an example of a person using boto, in case it's helpful (under Additional Info):

• BackBlaze B2 permission error treeverse/dvc#7372 (comment)

[ericmjl]

@machow thank you for chiming in! Yes, I am able to list files directly using fsspec. A minimally complex example is below:

import s3fs

fs = s3fs.S3FileSystem(
    endpoint_url=os.getenv("FSSPEC_S3_ENDPOINT_URL"),
    key=os.getenv("FSSPEC_S3_KEY"),
    secret=os.getenv("FSSPEC_S3_SECRET"),
)
fs.ls("pins-test")

This outputs:

['pins-test/.DS_Store', 'pins-test/._.DS_Store', 'pins-test/mtcars']

Which is correct! (The .DS_Store files come from me trying to mount them on my Mac.)

[isabelizimm]

Closing this issue as #237 was merged. Thank you again for the contribution! 👾