DataStore Tools

Integration utilities for MSQ DataStore/Prism platform — simplified access to Azure Fabric, Azure Blob/Data Lake storage, LLM APIs, and generic REST APIs.

This README is written to double as a reference for both humans and AI coding assistants (e.g. Claude Code). Every public method on every class is documented with its real signature, required environment variables, side effects, and known gotchas — pulled directly from the source, not just the happy path.

Installation

pip install data-store-tools

Configuration

Almost every class in this package loads secrets from a .env file in the current working directory (via load_env_file(".env"), a minimal parser in data_store_tools/tools/load_env.py — it is not python-dotenv; see Known issues & gotchas) and, for FabricTools, a datastore_config.yaml file.

.env keys by class

Key	Used by	Notes
OPENAI_API_KEY	LargeLanguageModelTools	Passed as api_key to the OpenAI SDK client
DEPLOYMENT	LargeLanguageModelTools	Model/deployment name, e.g. gpt-5-mini
ENDPOINT	LargeLanguageModelTools	Base URL; client is built with base_url=f"{ENDPOINT}/openai/v1/"
BLOB_STORAGE_CONNECTION_STRING	AzureTools	Required for every blob method (upload_to_blob, download_from_blob, blob_file_exists, etc.)
BLOB_CONTAINER_NAME	AzureTools	Required for every blob method
AZMLS_DATASTORE_NAME	AzureTools	Only needed for upload_to_ml_datastore
AZ_STORAGE_KEY, BLOB_STORAGE_ACCOUNT_NAME, AZ_TENANT_ID, AZ_CLIENT_ID	AzureTools	Read into self.* in __init__ but not currently used by any method — safe to leave unset
FABRIC_ACCOUNT_NAME	FabricTools	OneLake/ADLS account, e.g. onelake
AZ_TENANT_ID	FabricTools	Entra tenant ID, used for the SharePoint/Graph token
APP_CLIENT_ID, APP_CLIENT_SECRET	FabricTools	Service principal credentials for the SharePoint/Graph MSAL token. Storage/Fabric access itself goes through DefaultAzureCredential (az login / managed identity / env-based SP — whatever DefaultAzureCredential picks up), independent of these two

load_env_file silently does nothing if .env is missing, and any key it doesn't find is simply left unset (os.getenv(...) returns None) — there's no validation. A missing key usually surfaces later as a confusing network/auth error (e.g. a request to https://None.dfs.fabric.microsoft.com) rather than a clear "env var missing" message.

datastore_config.yaml

FabricTools requires this file in the working directory:

datastore:
  lakehouses:
    - "bronze"   # index 0
    - "silver"   # index 1
    - "gold"     # index 2

The three names are read positionally (config['datastore']['lakehouses'][0/1/2]), not by key — order matters. Separately, FabricTools hardcodes self.lakehouse_list = ['bronze', 'silver', 'gold', 'sandbox'] regardless of what's in the YAML, and it's this hardcoded list (not the config file) that create_datastore_table_name validates against — sandbox is always valid even though it's never in the config.

Known issues & gotchas

Read this before writing new code against the package — these are real behaviors in the current source, not hypotheticals:

• FabricTools() is not a cheap constructor. Instantiating it authenticates with DefaultAzureCredential, opens a DataLakeServiceClient, acquires an MSAL token for Microsoft Graph/SharePoint, calls list_accessible_lakehouses(), and validates the workspace name you passed — all in __init__. Expect it to be slow and to need working Azure auth and network access before you can do anything with it. It raises ValueError if FABRIC_WORKSPACE_NAME isn't one of the accessible workspaces.
• Stateful table targeting. create_datastore_table_name(lakehouse, data_subcat, data_contents) doesn't return a handle — it sets self.lakehouse / self.table_name on the instance as a side effect. load_table_to_lakehouse(df) then reads those instance attributes rather than taking lakehouse/table as arguments. You must call create_datastore_table_name before every load_table_to_lakehouse call, and one FabricTools instance can only "target" one table at a time.
• load_table_to_lakehouse mutates your DataFrame in place, adding load_date and load_dttm columns to the object you passed in (no copy is made).
• delete_table(..., confirm=False) blocks on input(). The default is an interactive yes/no prompt — always pass confirm=True in scripts, notebooks-run-non-interactively, or CI.
• The date_part parameter on create_datastore_table_name is dead code. It's accepted and defaults to today's date, but the returned table name (f"{data_subcat}_{data_contents}") never uses it.
• setup_logger() re-adds a file handler on every call. It calls logging.getLogger(__name__), which returns the same logger object every time within a process. If you instantiate more than one tool class (e.g. FabricTools and AzureTools) in the same script, each instantiation attaches another FileHandler, so every subsequent log line gets written to all the log files created so far, not just the newest one.
• APITools.make_call never raises on failure. It always returns a value (the parsed JSON, or None), and stashes status on the instance instead: check api.success / api.error_message after every call rather than wrapping it in try/except.
• AzureTools.ensure_blob_directory and create_blob_folder are functionally duplicates (one writes a .keep placeholder, the other .placeholder) — pick one per project for consistency.
• LargeLanguageModelTools branches request parameters on a hardcoded set of deployment name strings ({"gpt-4o-mini", "gpt-5-chat"} get legacy max_tokens/temperature; "gpt-5.4-nano" gets reasoning_effort="none"; anything else falls through to the max_completion_tokens-only branch). A new deployment name you haven't added to that logic will silently take the "else" path — check query_large_language_model's source if a new model needs special parameters.
• The version string in data_store_tools/__init__.py (__version__ = "1.1.0") is stale relative to the published package version in pyproject.toml (currently 2.4.2).

Quick Start

Azure Fabric Operations

from data_store_tools.fabric_tools import FabricTools

# project_name is only used as the log file prefix; FABRIC_WORKSPACE_NAME must be
# a workspace your identity can already see (validated during __init__)
fabric_tools = FabricTools("my_project_name", "MSQ")

# Target a table (sets self.lakehouse / self.table_name — see gotchas above)
lakehouse, table_name = fabric_tools.create_datastore_table_name(
    lakehouse="bronze", data_subcat="rtd", data_contents="raw"
)

# Upload a DataFrame to that table (adds load_date/load_dttm, appends by default)
fabric_tools.load_table_to_lakehouse(df, delete_old_table=False)

# Read a table back
df = fabric_tools.read_table_from_lakehouse("bronze", table_name)

# Read a file straight out of Files/ (csv, xlsx, sav)
df = fabric_tools.read_file_from_lakehouse("bronze", "raw/upload.csv")

Azure Storage Operations

from data_store_tools.azure_tools import AzureTools

azure = AzureTools("my_project_name")  # project_name is required — used for the logger

# Blob storage (CSV in / CSV out)
azure.upload_to_blob(df, "raw/my_data.csv")
df = azure.download_from_blob("raw/my_data.csv")

# Azure ML datastore upload (separate from the Fabric-based DataStore)
azure.upload_to_ml_datastore(df, blob_file_name="my_data.csv", blob_file_path="uploads")

API Integration

from data_store_tools.api_tools import APITools

api = APITools("my_project_name")
data = api.make_call("https://api.example.com/endpoint", params={"q": "test"})
if not api.success:
    print(api.error_message)

LLM Integration

from data_store_tools.large_language_tools import LargeLanguageModelTools

llm = LargeLanguageModelTools()  # reads OPENAI_API_KEY / DEPLOYMENT / ENDPOINT from .env
answer = llm.query_large_language_model(
    role_text="You are a helpful assistant.",
    content_text="What is 2+2?",
)

Full API Reference

FabricTools — data_store_tools/fabric_tools.py

FabricTools(project_name: str, FABRIC_WORKSPACE_NAME: str)

Sets up logging, loads .env and datastore_config.yaml, authenticates via DefaultAzureCredential + MSAL, opens the OneLake DataLakeServiceClient, and validates FABRIC_WORKSPACE_NAME against the accessible workspaces. See gotchas above — this does real work and real network calls.

Method	Signature	Behavior
change_workspace	(new_workspace_name: str)	Validates and switches self.FABRIC_WORKSPACE_NAME / self.file_system_client to a different workspace. Raises ValueError if not accessible.
refresh_token	()	Re-fetches the storage bearer token (self.token). Called automatically before writes in load_table_to_lakehouse.
check_a_datastore_table_exists	(lakehouse_name: str, table_name: str, schema: str = 'dbo') -> bool	Checks for {lakehouse}.Lakehouse/Tables/{schema}/{table_name}. schema is normalized with to_valid_name.
create_datastore_table_name	(lakehouse, data_subcat: str, data_contents: str, date_part=<today>) -> (lakehouse, table_name)	Validates lakehouse against self.lakehouse_list (bronze/silver/gold/sandbox) and sets self.lakehouse/self.table_name as a side effect. date_part is accepted but unused — see gotchas.
list_tables_in_lakehouse	(lakehouse_name: str) -> list[str]	Lists table dirs under Tables/, excluding any starting with _. Returns [] and prints on error rather than raising.
list_files_in_lakehouse	(lakehouse_name: str, folder: str = "", recursive: bool = True) -> list[str]	Lists file paths under Files/[folder]. Returns [] and prints on error rather than raising.
file_sha1_from_lakehouse	(path: str) -> str	Streams a file (chunked, not loaded fully into memory) and returns its SHA-1 hex digest.
delete_table	(lakehouse_name, table_name, schema='dbo', confirm=False) -> bool	Deletes the table's directory recursively. Blocks on an interactive prompt if confirm=False. Returns False if the table doesn't exist; re-raises on other errors.
validate_workspace	(workspace_name: str) -> None	Raises ValueError if not in self.workspaces. Called by __init__ and change_workspace.
load_table_to_lakehouse	(df: pd.DataFrame, delete_old_table=False, mode="append", schema='dbo')	Requires create_datastore_table_name to have been called first (reads self.lakehouse/self.table_name). Adds load_date/load_dttm columns to the passed-in df in place, converts to Arrow, refreshes the token, then write_deltalake with mode="overwrite" for a new table or the given mode for an existing one.
read_table_from_lakehouse	(lakehouse_name, table_name, schema='dbo') -> pd.DataFrame | None	Reads a Delta table via deltalake. Prints a warning and returns None (not an exception) if the table doesn't exist.
read_file_from_lakehouse	(lakehouse_name, file_name, raw_bytes=False)	Reads a file from Files/. raw_bytes=True → io.BytesIO. Otherwise dispatches on extension: .xlsx/.xls/.xlsm → pd.ExcelFile (call .parse(sheet_name) on it), .csv → pd.DataFrame, .sav → (df, meta) tuple via pyreadstat. Raises ValueError for anything else.
get_latest_file_from_lakehouse	(lakehouse_name, folder_path, file_extension=None)	Lists files under Files/{folder_path}, optionally filtered by extension, picks the most recently modified, and delegates to read_file_from_lakehouse. Raises FileNotFoundError if the folder/extension combo yields nothing.
load_sharepoint_excel	(file_relative_url: str, sheet_name=0) -> pd.DataFrame | dict[str, pd.DataFrame]	Downloads an Excel file from the hardcoded Forge-DataStore SharePoint site via Microsoft Graph, using the SP token acquired in __init__. Pass sheet_name=None to get a dict of all sheets. file_relative_url example: /sites/Forge-DataStore/Shared Documents/DataStore/00-MSQ/01-TRACE/TRACE operations master.xlsx.
list_accessible_lakehouses	() -> (list[dict], list[str])	Lists all file systems (lakehouses) the identity can see; prints them; returns (lakehouses, workspace_names). Called automatically in __init__.

Real end-to-end usage pattern (from the project's own dev notebook):

fabric_tools = FabricTools("data_store_tools_dev", "Lab")
fabric_tools.list_accessible_lakehouses()

lakehouse, table_name = fabric_tools.create_datastore_table_name("sandbox", "test_data", "random_numbers")
fabric_tools.load_table_to_lakehouse(df)
fabric_tools.check_a_datastore_table_exists(lakehouse, table_name)
fabric_tools.list_tables_in_lakehouse(lakehouse)

df_read = fabric_tools.read_table_from_lakehouse(lakehouse, table_name)
fabric_tools.delete_table(lakehouse, table_name, confirm=True)

AzureTools — data_store_tools/azure_tools.py

AzureTools(project_name: str)

Sets up logging and loads .env. Note the constructor requires project_name — it's used only for the log filename.

Method	Signature	Behavior
upload_to_ml_datastore	(data: pd.DataFrame, blob_file_name: str, blob_file_path: str)	Connects to the Azure ML workspace via Workspace.from_config() (needs a config.json findable by the AML SDK), gets the AZMLS_DATASTORE_NAME datastore, writes data to a temp CSV, and uploads via Dataset.File.upload_directory. This is a different system from the Fabric-based DataStore — don't confuse the two.
blob_file_exists	(blob_file_name: str) -> bool	True/False, swallows all exceptions as False.
ensure_blob_directory	(blob_directory: str)	Creates a zero-byte .keep placeholder blob under the prefix if nothing exists there yet. See gotchas re: duplicate of create_blob_folder.
upload_to_blob	(data: pd.DataFrame, blob_file_name: str)	Serializes data to CSV in memory and uploads, overwriting any existing blob of the same name.
upload_json_to_blob	(data, blob_file_name: str)	data is any JSON-serializable dict/list; uploads pretty-printed JSON, overwriting.
download_json_from_blob	(blob_file_name: str)	Returns the parsed JSON (dict/list).
download_from_blob	(blob_file_name: str, file_type: str = 'csv') -> pd.DataFrame	file_type must be 'csv' or 'json' (raises ValueError otherwise). Raises FileNotFoundError if the blob doesn't exist.
create_blob_folder	(folder_path: str)	Same idea as ensure_blob_directory, using a .placeholder file instead of .keep.

azure = AzureTools("my_project")
azure.create_blob_folder("data/processed/my_dataset")
azure.upload_to_blob(df, "data/processed/my_dataset/2026-08-21.csv")
azure.upload_json_to_blob({"rows": 42}, "data/processed/my_dataset/meta.json")

if azure.blob_file_exists("data/processed/my_dataset/2026-08-21.csv"):
    df = azure.download_from_blob("data/processed/my_dataset/2026-08-21.csv", file_type="csv")

meta = azure.download_json_from_blob("data/processed/my_dataset/meta.json")

APITools — data_store_tools/api_tools.py

APITools(project_name: str)

Thin, generic wrapper around requests.get with logging and normalized error handling — no auth, retry/backoff, or non-GET verbs are implemented despite the README's "Features" list describing broader ambitions; treat those as a roadmap, not current behavior.

Method	Signature	Behavior
make_call	(url: str, params: dict, timeout: int = 30)	Always returns a value, never raises — see gotchas. On success, returns the parsed JSON body. On any failure (non-200, timeout, connection error, invalid JSON, etc.), returns None and sets self.success = False / self.error_message.
api_response	(data, success: bool, error_message: str = None)	Called internally by make_call to log and stash state; you generally don't need to call this directly.

api = APITools("my_project")
data = api.make_call("https://api.example.com/items", params={"page": 1}, timeout=10)
if api.success:
    process(data)
else:
    logger.warning(f"API call failed: {api.error_message}")

LargeLanguageModelTools — data_store_tools/large_language_tools.py

LargeLanguageModelTools()

Reads OPENAI_API_KEY, DEPLOYMENT, ENDPOINT from .env and builds an openai.OpenAI client pointed at f"{ENDPOINT}/openai/v1/" — this works against any OpenAI-compatible gateway (including an Azure APIM front door), not the azure OpenAI SDK class.

Method	Signature	Behavior
query_large_language_model	(role_text="You are a helpful assistant.", content_text="What is 2+2?") -> str	Sends a single system+user turn (no conversation history, no streaming). Request parameters depend on self.LLM_DEPLOYMENT — see Known issues for the exact branching. Returns response.choices[0].message.content.

llm = LargeLanguageModelTools()
summary = llm.query_large_language_model(
    role_text="You are a terse data-quality assistant.",
    content_text=f"In one sentence, describe this dataframe: {df.describe().to_string()}",
)

Shared tooling — data_store_tools/tools/

utils.py:

Function	Signature	Behavior
setup_logger	(log_dir=None, log_prefix="amazon_reviews") -> logging.Logger	Every one of the four tool classes above calls this in __init__ with only log_prefix set (i.e. log_dir=None, which falls back to $AZUREML_OUTPUT_DIR or the system temp dir). Writes {log_dir}/{log_prefix}_{YYYYMMDD_HHMMSS}.log. See the file-handler-stacking gotcha above.
load_config	(config_path="config.yaml") -> dict	Plain yaml.safe_load. FabricTools calls this with "datastore_config.yaml".
to_valid_name	(schema: str) -> str	Normalizes a string into a safe lowercase schema/identifier name (spaces → _, strips non-alphanumerics, prefixes a leading digit with _). Used internally to sanitize the schema argument on Fabric table methods.
save_dataframe	(df, filepath)	df.to_excel(filepath), logs and swallows PermissionError.
load_or_create_dataframe	(filepath, columns=None) -> pd.DataFrame	Reads an existing .xlsx, or returns an empty DataFrame(columns=columns) if the file doesn't exist yet.
minmax	(column) -> pd.Series	Min-max scales a numeric column/Series onto roughly a 1–101 range.
check_how_many_api_calls_are_left	(SERP_API_KEY) -> int	Calls serpapi.com/account.json and returns plan_searches_left. Unrelated to the four main classes — a leftover helper for SerpAPI-based projects.

load_env.py:

Function	Signature	Behavior
load_env_file	(filepath, override_existing=False) -> None	Minimal, hand-rolled .env parser (not python-dotenv, despite python-dotenv being a listed dependency). Skips blank lines and # comments. Raises ValueError on a non-comment line with no =. By default (override_existing=False) it will not overwrite a key that's already set in os.environ, so real environment variables always win over the .env file. Returns silently (does nothing) if the file doesn't exist.

Architecture Compatibility

Designed for MSQ DataStore/Prism medallion architecture:

• Bronze Layer: Raw data ingestion
• Silver Layer: Cleaned and validated data
• Gold Layer: Production-ready, aggregated data
• Sandbox: Development and testing (not in datastore_config.yaml, but always accepted by FabricTools.create_datastore_table_name)

Cross-Tenant Support

Handles MSQ/Freemavens tenant scenarios:

• DefaultAzureCredential authentication (Fabric/OneLake storage)
• Service principal support via MSAL (SharePoint/Graph, in FabricTools)
• Workspace-specific operations (change_workspace, validate_workspace)
• OneLake path conventions ({lakehouse}.Lakehouse/Tables/..., {lakehouse}.Lakehouse/Files/...)

Project Organization

├── README.md               <- This file
├── Makefile                <- make requirements / lint / format / test / build / publish
├── pyproject.toml           <- Package metadata and dependencies (setuptools)
├── datastore_config.yaml    <- Lakehouse name config, read by FabricTools
├── .env                     <- Secrets/config, read by load_env_file (not committed)
├── data/
│   ├── external             <- Data from third-party sources
│   ├── interim               <- Intermediate, transformed data
│   ├── processed             <- Final, canonical data sets
│   └── raw                   <- Original, immutable data dumps
├── docs/                    <- mkdocs project; see docs/mkdocs.yml
├── notebooks/               <- Exploratory/dev notebooks
├── tests/                   <- pytest suite
└── data_store_tools/
    ├── __init__.py
    ├── api_tools.py          <- APITools
    ├── azure_tools.py        <- AzureTools
    ├── fabric_tools.py       <- FabricTools
    ├── large_language_tools.py <- LargeLanguageModelTools
    ├── tools/
    │   ├── load_env.py       <- load_env_file
    │   └── utils.py          <- setup_logger, load_config, to_valid_name, etc.
    └── legacy/                <- Older cookiecutter-style scaffolding (config/dataset/features/modeling/plots), not part of the public API described above

---

Contributing

Contributions welcome! See CONTRIBUTING.md for guidelines.

License

MIT License

Support

• Issues: https://github.com/freemavens/data_store_tools/issues
• Internal: Contact Forge Data Science team