Replies: 3 comments 7 replies
-
|
Facing the same issue as well… |
Beta Was this translation helpful? Give feedback.
0 replies
-
|
I have done a bit of playing and have learned the following things
Those two items led me to trying to inspect the available filesystems directly, where i was able to notice this warning message >>> from airflow import io
/home/lucas/code/telekom/gsus/pro-89/modeling/.venv/lib/python3.12/site-packages/airflow/configuration.py:859 DeprecationWarning: The secret_key option in [webserver] has been moved to the secret_key option in [api] - the old setting has been used, but please update your config.
>>> io._register_filesystems()
[2025-11-21T18:36:11.870+0100] {configuration.py:1077} WARNING - section/key [openlineage/namespace] not found in config
[2025-11-21T18:36:12.461+0100] {local.py:145} WARNING - To enable emitting Openlineage events, upgrade to Airflow 2.7 or install astronomer-cosmos[openlineage].
[2025-11-21T18:36:13.029+0100] {providers_manager.py:266} INFO - Optional provider feature disabled when importing 'airflow.providers.microsoft.azure.fs.msgraphfs.get_fs' from 'apache-airflow-providers-microsoft-azure' package
{'file': <function _file at 0x7f3f339f0a40>, 'local': <function _file at 0x7f3f339f0a40>, 'abfs': <function get_fs at 0x7f3f2eec3ce0>, 'abfss': <function get_fs at 0x7f3f2eec3ce0>, 'adl': <function get_fs at 0x7f3f2eec3ce0>}So I turned on debug logging and got some more details [2025-11-21T18:38:54.336+0100] {providers_manager.py:359} DEBUG - Initialization of Providers Manager[list] took 1.20 seconds
[2025-11-21T18:38:54.337+0100] {providers_manager.py:260} DEBUG - Optional feature disabled on exception when importing 'airflow.providers.microsoft.azure.fs.msgraphfs.get_fs' from 'apache-airflow-providers-microsoft-azure' package
Traceback (most recent call last):
File "/home/lucas/code/telekom/gsus/pro-89/modeling/.venv/lib/python3.12/site-packages/airflow/providers_manager.py", line 306, in _correctness_check
imported_class = import_string(class_name)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/lucas/code/telekom/gsus/pro-89/modeling/.venv/lib/python3.12/site-packages/airflow/utils/module_loading.py", line 39, in import_string
module = import_module(module_path)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/lucas/.local/share/uv/python/cpython-3.12.5-linux-x86_64-gnu/lib/python3.12/importlib/__init__.py", line 90, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "<frozen importlib._bootstrap>", line 1387, in _gcd_import
File "<frozen importlib._bootstrap>", line 1360, in _find_and_load
File "<frozen importlib._bootstrap>", line 1324, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'airflow.providers.microsoft.azure.fs.msgraphfs'
[2025-11-21T18:38:54.337+0100] {providers_manager.py:266} INFO - Optional provider feature disabled when importing 'airflow.providers.microsoft.azure.fs.msgraphfs.get_fs' from 'apache-airflow-providers-microsoft-azure' package
[2025-11-21T18:38:54.337+0100] {providers_manager.py:359} DEBUG - Initialization of Providers Manager[filesystems] took 1.20 seconds
This line in the provider info "filesystems": [
"airflow.providers.microsoft.azure.fs.adls",
"airflow.providers.microsoft.azure.fs.msgraphfs",
],should be "filesystems": [
"airflow.providers.microsoft.azure.fs.adls",
"airflow.providers.microsoft.azure.fs.msgraph",
],This discuss should be converted to a bug ticket and if it is ok, i will submit a patch. |
Beta Was this translation helpful? Give feedback.
2 replies
-
|
I created a quick PR #58568 |
Beta Was this translation helpful? Give feedback.
-
|
Thank you Lucas for the outstanding work on this. I did also try using 'msgraph' as a filesystem with the same result, so your findings about the incorrect naming to 'msgraphfs' makes sense to me. I would be very happy to see this patched. |
Beta Was this translation helpful? Give feedback.
-
|
Sadly, I don't think my PR #58568 is enough to make this functional. When I use the patch locally, it can find the filesystem but I get errors during the initialization.
At this point, I am a bit stuck. I might just have to use the |
Beta Was this translation helpful? Give feedback.
5 replies
-
|
I submitted too early. I added some more logging into def __init__(
self,
drive_id: str | None = None,
client_id: str | None = None,
tenant_id: str | None = None,
client_secret: str | None = None,
site_name: str | None = None,
drive_name: str | None = None,
oauth2_client_params: dict | None = None,
asynchronous: bool = False,
loop=None,
url_path: str | None = None,
**kwargs,
):
_logger.warning(f"MSGDriveFS drive_id={drive_id} client_id={client_id} tenant_id={tenant_id} client_secret={client_secret} oauth2_client_params={oauth2_client_params}")
# Get OAuth2 credentials from parameters or environment variables
# Check MSGRAPHFS_* variables first, then fall back to standard AZURE_* variables
self.client_id = (
client_id
or os.getenv("MSGRAPHFS_CLIENT_ID")
or os.getenv("AZURE_CLIENT_ID")
)
self.tenant_id = (
tenant_id
or os.getenv("MSGRAPHFS_TENANT_ID")
or os.getenv("AZURE_TENANT_ID")
)
self.client_secret = (
client_secret
or os.getenv("MSGRAPHFS_CLIENT_SECRET")
or os.getenv("AZURE_CLIENT_SECRET")
)
# Parse URL path if provided to extract site_name and drive_name
if url_path:
parsed_site, parsed_drive, _ = parse_msgraph_url(url_path)
# URL parameters override direct parameters
site_name = parsed_site or site_name
drive_name = parsed_drive or drive_name
# Determine operation mode (single-site if site and drive provided, OR drive_id provided)
self._multi_site_mode = not ((site_name and drive_name) or drive_id)
_logger.warning(f"MSGDriveFS initialized in multi_site_mode={self._multi_site_mode}")
# Set site_name and drive_name attributes for all modes
self.site_name = site_name
self.drive_name = drive_name
if self._multi_site_mode:
# Multi-site mode: cache for drive filesystem instances
self._drive_cache = {}
# Store credentials for creating drive instances
self._stored_client_id = self.client_id
self._stored_tenant_id = self.tenant_id
self._stored_client_secret = self.client_secret
self._stored_oauth2_client_params = oauth2_client_params
self._stored_kwargs = kwargs.copy()
# Build oauth2_client_params if not provided
if oauth2_client_params is None:
if not all([self.client_id, self.tenant_id, self.client_secret]):
raise ValueError(
"Either oauth2_client_params must be provided, or all of "
"client_id, tenant_id, and client_secret must be provided "
"(either as parameters or environment variables MSGRAPHFS_CLIENT_ID/"
"AZURE_CLIENT_ID, MSGRAPHFS_TENANT_ID/AZURE_TENANT_ID, "
"MSGRAPHFS_CLIENT_SECRET/AZURE_CLIENT_SECRET)"
)
# Build OAuth2 client parameters with proper configuration
oauth2_client_params = {
"client_id": self.client_id,
"client_secret": self.client_secret,
"token_endpoint": f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token",
"scope": " ".join(self.DEFAULT_SCOPES),
"grant_type": "client_credentials",
}
else:
# Extract credentials from provided params for later use
self.client_id = oauth2_client_params.get("client_id")
self.tenant_id = self._extract_tenant_from_token_endpoint(
oauth2_client_params.get("token_endpoint", "")
)
self.client_secret = oauth2_client_params.get("client_secret")
_logger.warning(f"MSGDriveFS oauth2_client_params={oauth2_client_params}")
super().__init__(
oauth2_client_params=oauth2_client_params,
asynchronous=asynchronous,
loop=loop,
**kwargs,
)
_logger.warning(f"MSGDriveFS post-super site_name={site_name} drive_name={drive_name} drive_id={drive_id}")
self.site_name = site_name
self.drive_name = drive_name
self.drive_id = drive_id
# We'll set the drive_url later if drive_id is determined
if self.drive_id:
self.drive_url = f"https://graph.microsoft.com/v1.0/drives/{self.drive_id}"
else:
self.drive_url = NoneNote that we clearly get the log from after the |
Beta Was this translation helpful? Give feedback.
-
|
Ok, i have confirmed that direct initialization of the filesystem does not do this double init and that it actually works as expected, here is my task definition where I attempt both import logging
from datetime import datetime
from airflow.providers.common.compat.sdk import BaseHook
from airflow.sdk import DAG, ObjectStoragePath, Variable, task
from msgraphfs import MSGDriveFS
logger = logging.getLogger(__name__)
@task
def list_sharepoint_files() -> list[dict[str, str]]:
"""
List files in SharePoint folder matching the configured pattern.
"""
drive_id = Variable.get("sharepoint_drive_id")
source_folder = Variable.get("sharepoint_source_folder")
file_pattern = Variable.get("sharepoint_file_pattern", default="*")
msgraph_conn_id = "sharepoint"
conn = BaseHook.get_connection(msgraph_conn_id)
fs = MSGDriveFS(
client_id=conn.login,
tenant_id=conn.host,
client_secret=conn.password,
# url_path=f"sharepoint://{drive_id}/{source_folder}/",
)
for p in fs.ls(f"/{drive_id}/{source_folder}/"):
logger.warning(f" - fs://{p}")
logger.warning("Using ObjectStoragePath to list files")
# Build SharePoint source path
source_path = ObjectStoragePath(
f"sharepoint://{msgraph_conn_id}/{drive_id}/{source_folder}/",
conn_id=msgraph_conn_id,
)
logger.info(f"Listing files in SharePoint: {source_path}")
logger.info(f"File pattern: {file_pattern}")
if not source_path.exists():
raise FileNotFoundError(f"SharePoint source folder does not exist: {source_path}")
matched_files = []
# Recursively search for matching files
for item in source_path.rglob(file_pattern):
if item.is_file():
# Store relative path from source folder
relative_path = item.relative_to(source_path)
matched_files.append(
{
"path": str(relative_path),
"name": item.name,
"full_path": str(item),
}
)
logger.info(f"Found {len(matched_files)} matching files")
for file_info in matched_files:
logger.info(f" - {file_info['path']}")
return matched_files
with DAG(
dag_id="sharepoint_to_blob",
description="Copy files from SharePoint to Azure Blob Storage",
schedule="@daily",
start_date=datetime(2025, 1, 1),
catchup=False,
is_paused_upon_creation=True,
tags=["sharepoint", "data-ingestion"],
) as dag:
files = list_sharepoint_files()And the resulting logs |
Beta Was this translation helpful? Give feedback.
-
|
I think this is likely an interaction between the ObjectStoragePath (and UPath, which it is built upon) and the URL we pass to it. For example, after looking into the how the Path is initialized, i updated my code to look like this logger.warning("Using ObjectStoragePath to list files")
# Build SharePoint source path
source_path = ObjectStoragePath(
f"sharepoint://{msgraph_conn_id}@{msgraph_conn_id}/{drive_id}/{source_folder}/",
)
logger.warning(f"Listing files in SharePoint: {source_path}")
logger.warning(f"File pattern: {file_pattern}")
if not source_path.exists():
raise FileNotFoundError(f"SharePoint source folder does not exist: {source_path}")I now get logs after the initialization of the Path, but still the same error |
Beta Was this translation helpful? Give feedback.
-
|
Is it possible to convert this to an issue/bug? Because this seems broken, not just a simple user / configuration issue. |
Beta Was this translation helpful? Give feedback.
-
|
Anyone can create an issue - and link to that dicussion. |
Beta Was this translation helpful? Give feedback.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
-
Hi all, I am attempting to ingest files from Sharepoint using the Microsoft Graph Filesystem included in apache-airflow-providers-microsoft-azure in accordance with this guide.
My DAG fails to import with this error:
Traceback (most recent call last): File "/home/user/run_airflow/.venv/lib64/python3.12/site-packages/airflow/sdk/io/store.py", line 63, in fs return get_fs(self.protocol, self.conn_id) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/user/run_airflow/.venv/lib64/python3.12/site-packages/airflow/sdk/io/fs.py", line 89, in get_fs raise ValueError(f"No filesystem registered for scheme {scheme}") from None ValueError: No filesystem registered for scheme sharepointI am unsure if this indicates my installation/configuration is incorrect or if this is a bug.
Airflow and Microsoft provider installed from PyPi/uv. Python version 3.12.
Beta Was this translation helpful? Give feedback.
All reactions