prefect.filesystems
¶
Azure
pydantic-model
¶
Store data as a file on Azure Datalake and Azure Blob Storage.
Examples:
Load stored Azure config:
from prefect.filesystems import Azure
az_block = Azure.load("BLOCK_NAME")
Source code in prefect/filesystems.py
class Azure(ReadableFileSystem, WritableFileSystem):
"""
Store data as a file on Azure Datalake and Azure Blob Storage.
Example:
Load stored Azure config:
```python
from prefect.filesystems import Azure
az_block = Azure.load("BLOCK_NAME")
```
"""
_block_type_name = "Azure"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/6AiQ6HRIft8TspZH7AfyZg/39fd82bdbb186db85560f688746c8cdd/azure.png?h=250"
bucket_path: str = Field(
...,
description="An Azure storage bucket path",
example="my-bucket/a-directory-within",
)
azure_storage_connection_string: Optional[SecretStr] = Field(
None,
title="Azure storage connection string",
description="Equivalent to the AZURE_STORAGE_CONNECTION_STRING environment variable",
)
azure_storage_account_name: Optional[SecretStr] = Field(
None,
title="Azure storage account name",
description="Equivalent to the AZURE_STORAGE_ACCOUNT_NAME environment variable",
)
azure_storage_account_key: Optional[SecretStr] = Field(
None,
title="Azure storage account key",
description="Equivalent to the AZURE_STORAGE_ACCOUNT_KEY environment variable",
)
_remote_file_system: RemoteFileSystem = None
@property
def basepath(self) -> str:
return f"az://{self.bucket_path}"
@property
def filesystem(self) -> RemoteFileSystem:
settings = {}
if self.azure_storage_connection_string:
settings[
"connection_string"
] = self.azure_storage_connection_string.get_secret_value()
if self.azure_storage_account_name:
settings[
"account_name"
] = self.azure_storage_account_name.get_secret_value()
if self.azure_storage_account_key:
settings["account_key"] = self.azure_storage_account_key.get_secret_value()
self._remote_file_system = RemoteFileSystem(
basepath=f"az://{self.bucket_path}", settings=settings
)
return self._remote_file_system
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> bytes:
"""
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
return await self.filesystem.get_directory(
from_path=from_path, local_path=local_path
)
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote directory.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
return await self.filesystem.put_directory(
local_path=local_path, to_path=to_path, ignore_file=ignore_file
)
async def read_path(self, path: str) -> bytes:
return await self.filesystem.read_path(path)
async def write_path(self, path: str, content: bytes) -> str:
return await self.filesystem.write_path(path=path, content=content)
azure_storage_account_key
pydantic-field
¶
Type: SecretStr
Equivalent to the AZURE_STORAGE_ACCOUNT_KEY environment variable
azure_storage_account_name
pydantic-field
¶
Type: SecretStr
Equivalent to the AZURE_STORAGE_ACCOUNT_NAME environment variable
azure_storage_connection_string
pydantic-field
¶
Type: SecretStr
Equivalent to the AZURE_STORAGE_CONNECTION_STRING environment variable
bucket_path
pydantic-field
required
¶
Type: str
An Azure storage bucket path
Azure.get_directory
async
¶
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
Source code in prefect/filesystems.py
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> bytes:
"""
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
return await self.filesystem.get_directory(
from_path=from_path, local_path=local_path
)
Azure.put_directory
async
¶
Uploads a directory from a given local path to a remote directory.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
Source code in prefect/filesystems.py
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote directory.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
return await self.filesystem.put_directory(
local_path=local_path, to_path=to_path, ignore_file=ignore_file
)
GCS
pydantic-model
¶
Store data as a file on Google Cloud Storage.
Examples:
Load stored GCS config:
from prefect.filesystems import GCS
gcs_block = GCS.load("BLOCK_NAME")
Source code in prefect/filesystems.py
class GCS(ReadableFileSystem, WritableFileSystem):
"""
Store data as a file on Google Cloud Storage.
Example:
Load stored GCS config:
```python
from prefect.filesystems import GCS
gcs_block = GCS.load("BLOCK_NAME")
```
"""
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/4CD4wwbiIKPkZDt4U3TEuW/c112fe85653da054b6d5334ef662bec4/gcp.png?h=250"
bucket_path: str = Field(
..., description="A GCS bucket path", example="my-bucket/a-directory-within"
)
service_account_info: Optional[SecretStr] = Field(
None, description="The contents of a service account keyfile as a JSON string."
)
project: Optional[str] = Field(
None,
description="The project the GCS bucket resides in. If not provided, the project will be inferred from the credentials or environment.",
)
@property
def basepath(self) -> str:
return f"gcs://{self.bucket_path}"
@property
def filesystem(self) -> RemoteFileSystem:
settings = {}
if self.service_account_info:
try:
settings["token"] = json.loads(
self.service_account_info.get_secret_value()
)
except json.JSONDecodeError:
raise ValueError(
"Unable to load provided service_account_info. Please make sure that the provided value is a valid JSON string."
)
remote_file_system = RemoteFileSystem(
basepath=f"gcs://{self.bucket_path}", settings=settings
)
return remote_file_system
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> bytes:
"""
Downloads a directory from a given remote path to a local directory.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
return await self.filesystem.get_directory(
from_path=from_path, local_path=local_path
)
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote directory.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
return await self.filesystem.put_directory(
local_path=local_path, to_path=to_path, ignore_file=ignore_file
)
async def read_path(self, path: str) -> bytes:
return await self.filesystem.read_path(path)
async def write_path(self, path: str, content: bytes) -> str:
return await self.filesystem.write_path(path=path, content=content)
bucket_path
pydantic-field
required
¶
Type: str
A GCS bucket path
project
pydantic-field
¶
Type: str
The project the GCS bucket resides in. If not provided, the project will be inferred from the credentials or environment.
service_account_info
pydantic-field
¶
Type: SecretStr
The contents of a service account keyfile as a JSON string.
GCS.get_directory
async
¶
Downloads a directory from a given remote path to a local directory.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
Source code in prefect/filesystems.py
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> bytes:
"""
Downloads a directory from a given remote path to a local directory.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
return await self.filesystem.get_directory(
from_path=from_path, local_path=local_path
)
GCS.put_directory
async
¶
Uploads a directory from a given local path to a remote directory.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
Source code in prefect/filesystems.py
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote directory.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
return await self.filesystem.put_directory(
local_path=local_path, to_path=to_path, ignore_file=ignore_file
)
LocalFileSystem
pydantic-model
¶
Store data as a file on a local file system.
Examples:
Load stored local file system config:
from prefect.filesystems import LocalFileSystem
local_file_system_block = LocalFileSystem.load("BLOCK_NAME")
Source code in prefect/filesystems.py
class LocalFileSystem(ReadableFileSystem, WritableFileSystem):
"""
Store data as a file on a local file system.
Example:
Load stored local file system config:
```python
from prefect.filesystems import LocalFileSystem
local_file_system_block = LocalFileSystem.load("BLOCK_NAME")
```
"""
_block_type_name = "Local File System"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/EVKjxM7fNyi4NGUSkeTEE/95c958c5dd5a56c59ea5033e919c1a63/image1.png?h=250"
basepath: Optional[str] = None
@validator("basepath", pre=True)
def cast_pathlib(cls, value):
if isinstance(value, Path):
return str(value)
return value
def _resolve_path(self, path: str) -> Path:
# Only resolve the base path at runtime, default to the current directory
basepath = (
Path(self.basepath).expanduser().resolve()
if self.basepath
else Path(".").resolve()
)
# Determine the path to access relative to the base path, ensuring that paths
# outside of the base path are off limits
path: Path = Path(path).expanduser()
if not path.is_absolute():
path = basepath / path
else:
path = path.resolve()
if not basepath in path.parents:
raise ValueError(
f"Attempted to write to path {path} outside of the base path {basepath}."
)
return path
async def get_directory(
self, from_path: str = None, local_path: str = None
) -> None:
"""
Copies a directory from one place to another on the local filesystem.
Defaults to copying the entire contents of the block's basepath to the current working directory.
"""
if from_path is None:
from_path = Path(self.basepath).expanduser()
if local_path is None:
local_path = Path(".").absolute()
if sys.version_info < (3, 8):
shutil.copytree(from_path, local_path)
else:
shutil.copytree(from_path, local_path, dirs_exist_ok=True)
async def _get_ignore_func(self, local_path: str, ignore_file: str):
with open(ignore_file, "r") as f:
ignore_patterns = f.readlines()
included_files = filter_files(local_path, ignore_patterns)
def ignore_func(directory, files):
return_val = [f for f in files if f not in included_files]
return return_val
return ignore_func
async def put_directory(
self, local_path: str = None, to_path: str = None, ignore_file: str = None
) -> None:
"""
Copies a directory from one place to another on the local filesystem.
Defaults to copying the entire contents of the current working directory to the block's basepath.
An `ignore_file` path may be provided that can include gitignore style expressions for filepaths to ignore.
"""
if to_path is None:
to_path = Path(self.basepath).expanduser()
if local_path is None:
local_path = Path(".").absolute()
if ignore_file:
ignore_func = await self._get_ignore_func(local_path, ignore_file)
else:
ignore_func = None
if local_path == to_path:
pass
else:
if sys.version_info < (3, 8):
shutil.copytree(local_path, to_path, ignore=ignore_func)
else:
shutil.copytree(
local_path, to_path, dirs_exist_ok=True, ignore=ignore_func
)
async def read_path(self, path: str) -> bytes:
path: Path = self._resolve_path(path)
# Check if the path exists
if not path.exists():
raise ValueError(f"Path {path} does not exist.")
# Validate that its a file
if not path.is_file():
raise ValueError(f"Path {path} is not a file.")
async with await anyio.open_file(str(path), mode="rb") as f:
content = await f.read()
return content
async def write_path(self, path: str, content: bytes) -> str:
path: Path = self._resolve_path(path)
# Construct the path if it does not exist
path.parent.mkdir(exist_ok=True, parents=True)
# Check if the file already exists
if path.exists() and not path.is_file():
raise ValueError(f"Path {path} already exists and is not a file.")
async with await anyio.open_file(path, mode="wb") as f:
await f.write(content)
LocalFileSystem.get_directory
async
¶
Copies a directory from one place to another on the local filesystem.
Defaults to copying the entire contents of the block's basepath to the current working directory.
Source code in prefect/filesystems.py
async def get_directory(
self, from_path: str = None, local_path: str = None
) -> None:
"""
Copies a directory from one place to another on the local filesystem.
Defaults to copying the entire contents of the block's basepath to the current working directory.
"""
if from_path is None:
from_path = Path(self.basepath).expanduser()
if local_path is None:
local_path = Path(".").absolute()
if sys.version_info < (3, 8):
shutil.copytree(from_path, local_path)
else:
shutil.copytree(from_path, local_path, dirs_exist_ok=True)
LocalFileSystem.put_directory
async
¶
Copies a directory from one place to another on the local filesystem.
Defaults to copying the entire contents of the current working directory to the block's basepath.
An ignore_file path may be provided that can include gitignore style expressions for filepaths to ignore.
Source code in prefect/filesystems.py
async def put_directory(
self, local_path: str = None, to_path: str = None, ignore_file: str = None
) -> None:
"""
Copies a directory from one place to another on the local filesystem.
Defaults to copying the entire contents of the current working directory to the block's basepath.
An `ignore_file` path may be provided that can include gitignore style expressions for filepaths to ignore.
"""
if to_path is None:
to_path = Path(self.basepath).expanduser()
if local_path is None:
local_path = Path(".").absolute()
if ignore_file:
ignore_func = await self._get_ignore_func(local_path, ignore_file)
else:
ignore_func = None
if local_path == to_path:
pass
else:
if sys.version_info < (3, 8):
shutil.copytree(local_path, to_path, ignore=ignore_func)
else:
shutil.copytree(
local_path, to_path, dirs_exist_ok=True, ignore=ignore_func
)
RemoteFileSystem
pydantic-model
¶
Store data as a file on a remote file system.
Supports any remote file system supported by fsspec. The file system is specified
using a protocol. For example, "s3://my-bucket/my-folder/" will use S3.
Examples:
Load stored remote file system config:
from prefect.filesystems import RemoteFileSystem
remote_file_system_block = RemoteFileSystem.load("BLOCK_NAME")
Source code in prefect/filesystems.py
class RemoteFileSystem(ReadableFileSystem, WritableFileSystem):
"""
Store data as a file on a remote file system.
Supports any remote file system supported by `fsspec`. The file system is specified
using a protocol. For example, "s3://my-bucket/my-folder/" will use S3.
Example:
Load stored remote file system config:
```python
from prefect.filesystems import RemoteFileSystem
remote_file_system_block = RemoteFileSystem.load("BLOCK_NAME")
```
"""
_block_type_name = "Remote File System"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/4CxjycqILlT9S9YchI7o1q/ee62e2089dfceb19072245c62f0c69d2/image12.png?h=250"
basepath: str
settings: dict = Field(default_factory=dict)
# Cache for the configured fsspec file system used for access
_filesystem: fsspec.AbstractFileSystem = None
@validator("basepath")
def check_basepath(cls, value):
scheme, netloc, _, _, _ = urllib.parse.urlsplit(value)
if not scheme:
raise ValueError(f"Base path must start with a scheme. Got {value!r}.")
if not netloc:
raise ValueError(
f"Base path must include a location after the scheme. Got {value!r}."
)
if scheme == "file":
raise ValueError(
"Base path scheme cannot be 'file'. Use `LocalFileSystem` instead for local file access."
)
return value
def _resolve_path(self, path: str) -> str:
base_scheme, base_netloc, base_urlpath, _, _ = urllib.parse.urlsplit(
self.basepath
)
scheme, netloc, urlpath, _, _ = urllib.parse.urlsplit(path)
# Confirm that absolute paths are valid
if scheme:
if scheme != base_scheme:
raise ValueError(
f"Path {path!r} with scheme {scheme!r} must use the same scheme as the base path {base_scheme!r}."
)
if netloc:
if (netloc != base_netloc) or not urlpath.startswith(base_urlpath):
raise ValueError(
f"Path {path!r} is outside of the base path {self.basepath!r}."
)
return f"{self.basepath.rstrip('/')}/{urlpath.lstrip('/')}"
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> None:
"""
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
if from_path is None:
from_path = str(self.basepath)
if local_path is None:
local_path = Path(".").absolute()
return self.filesystem.get(from_path, local_path, recursive=True)
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote direcotry.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
if to_path is None:
to_path = str(self.basepath)
if local_path is None:
local_path = "."
included_files = None
if ignore_file:
with open(ignore_file, "r") as f:
ignore_patterns = f.readlines()
included_files = filter_files(local_path, ignore_patterns)
counter = 0
for f in glob.glob("**", recursive=True):
if ignore_file and f not in included_files:
continue
if to_path.endswith("/"):
fpath = to_path + f
else:
fpath = to_path + "/" + f
if Path(f).is_dir():
pass
else:
self.filesystem.put_file(f, fpath, overwrite=True)
counter += 1
return counter
async def read_path(self, path: str) -> bytes:
path = self._resolve_path(path)
with self.filesystem.open(path, "rb") as file:
content = await run_sync_in_worker_thread(file.read)
return content
async def write_path(self, path: str, content: bytes) -> str:
path = self._resolve_path(path)
dirpath = path[: path.rindex("/")]
self.filesystem.makedirs(dirpath, exist_ok=True)
with self.filesystem.open(path, "wb") as file:
await run_sync_in_worker_thread(file.write, content)
@property
def filesystem(self) -> fsspec.AbstractFileSystem:
if not self._filesystem:
scheme, _, _, _, _ = urllib.parse.urlsplit(self.basepath)
try:
self._filesystem = fsspec.filesystem(scheme, **self.settings)
except ImportError as exc:
# The path is a remote file system that uses a lib that is not installed
raise RuntimeError(
f"File system created with scheme {scheme!r} from base path "
f"{self.basepath!r} could not be created. "
"You are likely missing a Python module required to use the given "
"storage protocol."
) from exc
return self._filesystem
RemoteFileSystem.get_directory
async
¶
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
Source code in prefect/filesystems.py
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> None:
"""
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
if from_path is None:
from_path = str(self.basepath)
if local_path is None:
local_path = Path(".").absolute()
return self.filesystem.get(from_path, local_path, recursive=True)
RemoteFileSystem.put_directory
async
¶
Uploads a directory from a given local path to a remote direcotry.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
Source code in prefect/filesystems.py
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote direcotry.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
if to_path is None:
to_path = str(self.basepath)
if local_path is None:
local_path = "."
included_files = None
if ignore_file:
with open(ignore_file, "r") as f:
ignore_patterns = f.readlines()
included_files = filter_files(local_path, ignore_patterns)
counter = 0
for f in glob.glob("**", recursive=True):
if ignore_file and f not in included_files:
continue
if to_path.endswith("/"):
fpath = to_path + f
else:
fpath = to_path + "/" + f
if Path(f).is_dir():
pass
else:
self.filesystem.put_file(f, fpath, overwrite=True)
counter += 1
return counter
S3
pydantic-model
¶
Store data as a file on AWS S3.
Examples:
Load stored S3 config:
from prefect.filesystems import S3
s3_block = S3.load("BLOCK_NAME")
Source code in prefect/filesystems.py
class S3(ReadableFileSystem, WritableFileSystem):
"""
Store data as a file on AWS S3.
Example:
Load stored S3 config:
```python
from prefect.filesystems import S3
s3_block = S3.load("BLOCK_NAME")
```
"""
_block_type_name = "S3"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/1jbV4lceHOjGgunX15lUwT/db88e184d727f721575aeb054a37e277/aws.png?h=250"
bucket_path: str = Field(
..., description="An S3 bucket path", example="my-bucket/a-directory-within"
)
aws_access_key_id: SecretStr = Field(None, title="AWS Access Key ID")
aws_secret_access_key: SecretStr = Field(None, title="AWS Secret Access Key")
_remote_file_system: RemoteFileSystem = None
@property
def basepath(self) -> str:
return f"s3://{self.bucket_path}"
@property
def filesystem(self) -> RemoteFileSystem:
settings = {}
if self.aws_access_key_id:
settings["key"] = self.aws_access_key_id.get_secret_value()
if self.aws_secret_access_key:
settings["secret"] = self.aws_secret_access_key.get_secret_value()
self._remote_file_system = RemoteFileSystem(
basepath=f"s3://{self.bucket_path}", settings=settings
)
return self._remote_file_system
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> bytes:
"""
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
return await self.filesystem.get_directory(
from_path=from_path, local_path=local_path
)
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote direcotry.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
return await self.filesystem.put_directory(
local_path=local_path, to_path=to_path, ignore_file=ignore_file
)
async def read_path(self, path: str) -> bytes:
return await self.filesystem.read_path(path)
async def write_path(self, path: str, content: bytes) -> str:
return await self.filesystem.write_path(path=path, content=content)
bucket_path
pydantic-field
required
¶
Type: str
An S3 bucket path
S3.get_directory
async
¶
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
Source code in prefect/filesystems.py
async def get_directory(
self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> bytes:
"""
Downloads a directory from a given remote path to a local direcotry.
Defaults to downloading the entire contents of the block's basepath to the current working directory.
"""
return await self.filesystem.get_directory(
from_path=from_path, local_path=local_path
)
S3.put_directory
async
¶
Uploads a directory from a given local path to a remote direcotry.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
Source code in prefect/filesystems.py
async def put_directory(
self,
local_path: Optional[str] = None,
to_path: Optional[str] = None,
ignore_file: Optional[str] = None,
) -> int:
"""
Uploads a directory from a given local path to a remote direcotry.
Defaults to uploading the entire contents of the current working directory to the block's basepath.
"""
return await self.filesystem.put_directory(
local_path=local_path, to_path=to_path, ignore_file=ignore_file
)