Power Query (Get & Transform) — the queries a workbook is fed by.
Data that arrives through a query has no formula behind it. The cells hold
values, the calculation engine sees a plain range, and the thing that filled
it — a CSV on a share, a database, another table of the same workbook — lives
in a part of the package no spreadsheet engine ever opens. Read it back and
the range stops being a dead end.
Where it sits in the file: a customXml item whose schema is
http://schemas.microsoft.com/DataMashup holds a UTF-16 document whose one
element is base64. Decoded, that is a uint32 version, a uint32 length,
then a ZIP whose Formulas/Section1.m carries the M source of every query
in plain text. xl/connections.xml names the query each connection loads,
and xl/queryTables/*.xml ties that connection to the table on a sheet — so
the landing range is reachable too.
What is read out of the M is deliberately shallow: the sources a query names
and the queries it chains from. M is a full language and this is not an
interpreter; a source built at run time is invisible here exactly as a VBA
range built at run time is invisible to the VBA reader.
QuerySource
dataclass
One thing a query reads: a table, another query, a file, a server.
Source code in src/linexcel/powerquery.py
| @dataclass(frozen=True)
class QuerySource:
"""One thing a query reads: a table, another query, a file, a server."""
kind: str
target: str
function: str
def as_dict(self) -> dict[str, str]:
return {"kind": self.kind, "target": self.target, "function": self.function}
|
Destination
dataclass
Where a query lands: the sheet, the range, the table Excel created.
Source code in src/linexcel/powerquery.py
| @dataclass(frozen=True)
class Destination:
"""Where a query lands: the sheet, the range, the table Excel created."""
sheet: str | None
ref: str | None
table: str | None
def as_dict(self) -> dict[str, str | None]:
return {"sheet": self.sheet, "ref": self.ref, "table": self.table}
|
Query
dataclass
One Power Query query: its M source, what it reads, where it lands.
Source code in src/linexcel/powerquery.py
| @dataclass
class Query:
"""One Power Query query: its M source, what it reads, where it lands."""
name: str
source: str
sources: list[QuerySource] = field(default_factory=list)
loaded_to: list[Destination] = field(default_factory=list)
@property
def loaded(self) -> bool:
"""``False`` for a connection-only query — computed, never written."""
return bool(self.loaded_to)
def outside_sources(self) -> list[QuerySource]:
"""The sources that are not in this workbook, so not in the graph."""
return [s for s in self.sources if s.kind not in ("table", "query")]
|
loaded
property
False for a connection-only query — computed, never written.
outside_sources
outside_sources() -> list[QuerySource]
The sources that are not in this workbook, so not in the graph.
Source code in src/linexcel/powerquery.py
| def outside_sources(self) -> list[QuerySource]:
"""The sources that are not in this workbook, so not in the graph."""
return [s for s in self.sources if s.kind not in ("table", "query")]
|
read_queries
read_queries(data: bytes) -> list[Query]
Every query of a workbook, in the order the mashup declares them.
An empty list means what it says: no Power Query in the file, or a mashup
part this reader could not make sense of. Either way the analysis carries
on — a query that cannot be read costs the graph a node, not a run.
Source code in src/linexcel/powerquery.py
| def read_queries(data: bytes) -> list[Query]:
"""Every query of a workbook, in the order the mashup declares them.
An empty list means what it says: no Power Query in the file, or a mashup
part this reader could not make sense of. Either way the analysis carries
on — a query that cannot be read costs the graph a node, not a run.
"""
section = read_section(data)
if not section:
return []
members = parse_section(section)
if not members:
return []
destinations = read_destinations(data)
names = set(members)
return [
Query(
name=name,
source=body,
sources=scan_sources(body, names - {name}),
loaded_to=destinations.get(name.casefold(), []),
)
for name, body in members.items()
]
|
query_warning
query_warning(queries: list[Query]) -> str | None
One line for the queries that feed the workbook from outside it.
A query whose source is a file, a URL or a server is a dependency of the
same nature as a link to another workbook: the values it produced are in
the file, what produced them is not. The graph shows the query and names
its source; nobody should read that as the source having been checked.
Source code in src/linexcel/powerquery.py
| def query_warning(queries: list[Query]) -> str | None:
"""One line for the queries that feed the workbook from outside it.
A query whose source is a file, a URL or a server is a dependency of the
same nature as a link to another workbook: the values it produced are in
the file, what produced them is not. The graph shows the query and names
its source; nobody should read that as the source having been checked.
"""
if not queries:
return None
loaded = sum(1 for query in queries if query.loaded)
if len(queries) == 1:
head = "1 Power Query query feeds this workbook" + (
", loaded onto a sheet."
if loaded
else ", loaded nowhere (connection only)."
)
else:
head = (
f"{len(queries)} Power Query queries feed this workbook, "
f"{loaded} of them loaded onto a sheet."
)
parts = [head]
outside = sorted(
{source.target for query in queries for source in query.outside_sources()}
)
if outside:
shown = ", ".join(outside[:MAX_QUERY_SOURCES_SHOWN])
if len(outside) > MAX_QUERY_SOURCES_SHOWN:
shown += f", … (+{len(outside) - MAX_QUERY_SOURCES_SHOWN})"
parts.append(
f"Their data comes from outside the file and was not read: {shown}."
)
return " ".join(parts)
|
read_section
read_section(data: bytes) -> str | None
The Section1.m text of the workbook's mashup, if it has one.
Source code in src/linexcel/powerquery.py
| def read_section(data: bytes) -> str | None:
"""The ``Section1.m`` text of the workbook's mashup, if it has one."""
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
for part in zf.namelist():
if not re.fullmatch(r"customXml/item\d+\.xml", part):
continue
root = _mashup_root(zf.read(part))
if root is None:
continue
section = _unpack_mashup(root.text or "")
if section is not None:
return section
except Exception:
return None
return None
|
parse_section
parse_section(text: str) -> dict[str, str]
section Section1; shared X = …; → {name: M source}.
parse_section('section Section1;\nshared Sales = let x = 1 in x;')
Source code in src/linexcel/powerquery.py
| def parse_section(text: str) -> dict[str, str]:
"""``section Section1; shared X = …;`` → ``{name: M source}``.
>>> parse_section('section Section1;\\nshared Sales = let x = 1 in x;')
{'Sales': 'let x = 1 in x'}
"""
out: dict[str, str] = {}
for statement in _statements(text):
head = _MEMBER_RE.match(statement)
if head is None:
continue
name = _identifier(head.group(1), head.group(2))
body = statement[head.end() :].strip()
if name and body:
out[name] = body
return out
|
scan_sources
scan_sources(
m_source: str, other_queries: set[str]
) -> list[QuerySource]
What one query reads, read off its M source.
Three kinds of answer: a table or named range of this workbook, another
query of the same file, or something outside both — a path, a URL, a
server. Anything an expression computes rather than spells out is not
here, and no static reader can put it there.
Source code in src/linexcel/powerquery.py
| def scan_sources(m_source: str, other_queries: set[str]) -> list[QuerySource]:
"""What one query reads, read off its M source.
Three kinds of answer: a table or named range of this workbook, another
query of the same file, or something outside both — a path, a URL, a
server. Anything an expression computes rather than spells out is not
here, and no static reader can put it there.
"""
found: list[QuerySource] = []
for match in _CURRENT_WORKBOOK_RE.finditer(m_source):
found.append(
QuerySource("table", _unquote(match.group(1)), "Excel.CurrentWorkbook")
)
for match in _CALL_RE.finditer(m_source):
function, target = match.group(1), _unquote(match.group(2))
kind = CONNECTORS.get(function)
if kind is None and function.rpartition(".")[2].endswith(DATABASE_SUFFIXES):
kind = "database"
if kind is not None and target:
found.append(QuerySource(kind, target, function))
for name in _referenced_queries(m_source, other_queries):
found.append(QuerySource("query", name, "let"))
seen: set[tuple[str, str]] = set()
unique: list[QuerySource] = []
for source in found:
key = (source.kind, source.target)
if key not in seen:
seen.add(key)
unique.append(source)
return unique
|
read_destinations
read_destinations(
data: bytes,
) -> dict[str, list[Destination]]
{query name, casefolded: [where it loads]}.
Follows the chain the format itself uses: worksheet → table part →
query table → connection → the query the connection names. A query that
loads nowhere — connection only, or straight into the data model — has no
entry, which is the difference the report shows.
Source code in src/linexcel/powerquery.py
| def read_destinations(data: bytes) -> dict[str, list[Destination]]:
"""``{query name, casefolded: [where it loads]}``.
Follows the chain the format itself uses: worksheet → table part →
query table → connection → the query the connection names. A query that
loads nowhere — connection only, or straight into the data model — has no
entry, which is the difference the report shows.
"""
out: dict[str, list[Destination]] = defaultdict(list)
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
connections = _connection_queries(zf)
if not connections:
return {}
book = _xml(zf, "xl/workbook.xml")
if book is None:
return {}
rels = _rels(zf, "xl/workbook.xml")
for element in book.iter():
if _tag(element) != "sheet":
continue
sheet = element.get("name")
rid = next(
(v for k, v in element.attrib.items() if _tag_name(k) == "id"), None
)
part = rels.get(rid or "")
if not sheet or not part:
continue
for table_part in _rels(zf, part, kind="table").values():
table = _xml(zf, table_part)
if table is None:
continue
for qt_part in _rels(zf, table_part, kind="queryTable").values():
query_table = _xml(zf, qt_part)
if query_table is None:
continue
query = connections.get(query_table.get("connectionId") or "")
if not query:
continue
out[query.casefold()].append(
Destination(
sheet=sheet,
ref=table.get("ref"),
table=table.get("displayName") or table.get("name"),
)
)
except Exception:
return {}
return dict(out)
|