Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .flake8
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
[flake8]
extend-ignore = E203
line-length = 79
28 changes: 28 additions & 0 deletions src/docker_volume_analyzer/docker_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,3 +94,31 @@ def remove_volume(self, volume_name: str) -> None:
raise docker.errors.APIError(
f"Failed to remove volume '{volume_name}': {e}"
) from e

def get_directory_informations_with_find(
self, volume_name: str, directory: str | None = None
) -> Union[str, None]:
"""
Gets directory information using 'find' command.

Args:
volume_name (str): Docker volume name.
directory (str): Directory path inside the volume.

Returns:
str | None: Output of the 'find' with stat command
or None if failed.
"""
path = (
f"/mnt/docker_volume/{directory}"
if directory
else "/mnt/docker_volume"
)

command = [
"sh",
"-c",
f"find {path} -exec stat -c '%F|%n|%s|%A|%U|%G|%Y' {{}} \\;",
]
output = self._run_in_container(command, volume_name)
return output if output else None
147 changes: 147 additions & 0 deletions src/docker_volume_analyzer/filesystem.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Optional


@dataclass
class FileNode:
"""
Represents a file or directory in the file system.

Attributes:
name (str): The name of the file or directory.
path (str): The full path of the file or directory.
size (int): The size of the file in bytes.
mtime (datetime): The last modified time of the file or directory.
mode (str): The file mode (permissions) as a string.
user (str): The owner of the file or directory.
group (str): The group owner of the file or directory.
is_directory (bool): True if the node is a directory,
False if it is a file.
childrens (Dict[str, "FileNode"]): A dictionary of child nodes
where the key is the child's name.
parent (Optional["FileNode"]): A reference to the parent node
or None if it is the root.
"""

name: str
path: str
size: int
mtime: datetime
mode: str
user: str
group: str
is_directory: bool
childrens: Dict[str, "FileNode"] = field(default_factory=dict)
parent: Optional["FileNode"] = None


class FileSystem:
"""
Represents a file system structure, allowing for the addition
and retrieval of file nodes.

Attributes:
root (FileNode): The root node of the file system.
index (Dict[str, FileNode]): A dictionary mapping relative
paths to their corresponding FileNode.
"""

def __init__(self):
self.root = FileNode(
name="",
path="",
size=0,
mtime=datetime.now(),
mode="",
user="",
group="",
is_directory=True,
)
self.index = {"": self.root}

def add_node(self, node: FileNode):
parts = node.path.strip("/").split("/")
current = self.root
full_path = ""

for i, part in enumerate(parts):
full_path = "/".join(parts[: i + 1])
if full_path not in self.index:
is_last = i == len(parts) - 1
n = FileNode(
name=part,
path=full_path,
size=node.size if is_last else 0,
mtime=node.mtime,
mode=node.mode,
user=node.user,
group=node.group,
is_directory=node.is_directory if is_last else True,
parent=current,
)
current.childrens[part] = n
self.index[full_path] = n
current = self.index[full_path]

def compute_directory_sizes(self) -> "FileSystem":
"""
Compute the total size of each directory by summing the sizes
of its files, subdirectories,
and the directory's own size (e.g., 4 KB for metadata).
"""
for path in sorted(
self.index.keys(), key=lambda x: x.count("/"), reverse=True
):
node = self.index[path]
if node.is_directory:
total_size = node.size

for child in node.childrens.values():
total_size += child.size

node.size = total_size

return self


def parse_find_output(
output: str, strip_prefix: str = "/mnt/docker_volume"
) -> FileSystem:
"""
Parses the output of the 'find' command with stat
and builds a FileSystem object.

Args:
output (str): The output string from the 'find' command,
strip_prefix (str): A prefix to strip from the path in the output
default is '/mnt/docker_volume'.

Returns:
FileSystem: An instance of FileSystem containing the parsed file nodes.
"""
fs = FileSystem()
for line in output.strip().split("\n"):
try:
type_str, path, size, mode, user, group, mtime = line.split("|")

path = (
path[len(strip_prefix) :].lstrip("/")
if path.startswith(strip_prefix)
else path.lstrip("/")
)

node = FileNode(
name=path.split("/")[-1],
path=path,
size=int(size),
mtime=datetime.fromtimestamp(int(mtime)),
mode=mode,
user=user,
group=group,
is_directory=(type_str == "directory"),
)
fs.add_node(node)
except Exception as e:
print(f"Skipping malformed line: {line} ({e})")
return fs
134 changes: 128 additions & 6 deletions src/docker_volume_analyzer/tui.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import os

from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Container, Horizontal, Vertical
from textual.events import Key
from textual.screen import ModalScreen
from textual.widgets import Button, DataTable, Footer, Header, Static

Expand All @@ -19,6 +21,7 @@ class DockerTUI(App):
("ctrl+q", "quit", "Quit"),
("i", "information", "Show information"),
("d", "delete_volume", "Delete volume"),
("b", "browse", "Browse volume"),
]
CSS_PATH = "tui.tcss"

Expand Down Expand Up @@ -139,6 +142,18 @@ def delete_callback(confirmed: bool) -> None:
)
)

def action_browse(self):
"""
An action to browse the contents of the selected volume.
"""

table = self.query_one(DataTable)
selected_row = table.cursor_row

volume_row = table.get_row_at(selected_row)

self.push_screen(VolumeBrowserScreen(self.app.manager, volume_row[0]))


class VolumeDetailScreen(ModalScreen):
"""
Expand Down Expand Up @@ -242,19 +257,126 @@ def __init__(self, message: str, callback):
self.callback = callback

def compose(self) -> ComposeResult:
yield Header()
with Container(id="dialog"):
yield Static(self.message, classes="question")
with Horizontal(classes="buttons"):
yield Button("No", variant="error", id="no_button")
yield Button("Yes", variant="success", id="yes_button")
yield Footer()

def on_button_pressed(self, event):
if event.button.id == "yes_button":
self.callback(True)
else:
self.callback(False)
self.callback(event.button.id == "yes_button")


class VolumeBrowserScreen(ModalScreen):
"""
A modal screen to browse the contents of a Docker volume.
"""

BINDINGS = [
("b", "back", "Back"),
Binding("enter", "select_cursor", "Select", show=True),
]

ICON_DIRECTORY = "📁 "
ICON_FILE = "📄 "

def __init__(self, volume_manager: VolumeManager, volume_name: str):
super().__init__()
self.volume_name = volume_name
self.volume_manager = volume_manager
self.volume_tree = self.volume_manager.get_volume_tree(volume_name)
self.current_path = ""

def compose(self) -> ComposeResult:
with Container(id="dialog"):

yield Static(
f"[b]Browsing volume:[/b] {self.volume_name}",
classes="title",
)
yield Static(
f"[b]Current path:[/b] {self.current_path}",
id="current_path",
)
yield DataTable(
id="file_tree",
cursor_type="row",
show_cursor=True,
classes="file-tree",
zebra_stripes=True,
)
with Container(id="shortcuts", classes="shortcuts-container"):
with Horizontal(classes="shortcuts-list"):
yield Static("[b]Enter:[/b]", classes="shortcut shortcut-key")
yield Static(
"Open selected directory", classes="shortcut shortcut-desc"
)
yield Static(
"[b]Backspace:[/b]", classes="shortcut shortcut-key"
)
yield Static(
"Open parent directory", classes="shortcut shortcut-desc"
)

def on_mount(self) -> None:
"""Load the file tree when the screen is mounted."""
table = self.query_one(DataTable)
table.add_columns("", "Name", "Size", "Last Modified")
self.load_data()

def load_data(self) -> None:

table = self.query_one(DataTable)
table.clear()
directory_informations = self.volume_tree.index.get(
self.current_path, {}
)
if not directory_informations:
self.query_one(DataTable).add_row(
"No files found in this directory."
)
return ()

for name, node in directory_informations.childrens.items():
table.add_row(
(
f"{self.ICON_DIRECTORY}"
if node.is_directory
else f"{self.ICON_FILE}"
),
name,
f"{node.size} bytes",
node.mtime.strftime("%Y-%m-%d %H:%M:%S"),
)

self.query_one("#current_path").update(
f"[b]Current path:[/b] {self.current_path}/"
)

def action_back(self) -> None:
"""An action to go back to the previous screen."""
self.app.pop_screen()
self.app.refresh()

def on_key(self, event: Key) -> None:
table = self.query_one(DataTable)
if event.key == "enter":
selected = table.cursor_row
row_data = table.get_row_at(selected)
selected_name = row_data[1]
selected_node = self.volume_tree.index.get(
self.current_path, {}
).childrens.get(selected_name)

if selected_node and selected_node.is_directory:
self.current_path = (
f"{self.current_path}/{selected_name}".strip("/")
)
self.load_data()
elif event.key == "backspace":
self.current_path = os.path.dirname(self.current_path)
table = self.query_one(DataTable)
self.load_data()


if __name__ == "__main__": # pragma: no cover
Expand Down
28 changes: 27 additions & 1 deletion src/docker_volume_analyzer/tui.tcss
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ Screen {
}

#dialog {
height: 70%;
height: 90%;
background: $panel;
color: $text;
border: tall $background;
Expand Down Expand Up @@ -55,3 +55,29 @@ Button {
height: auto;
color: white;
}

#file_tree{
height: 100%;
}


/* Liste des raccourcis en ligne */
.shortcuts-list {
align: left middle;
padding: 0 1;
}

/* Élément individuel de raccourci */
.shortcut-desc {
color: $footer-description-foreground;
padding-left: 1;
padding-right: 2;
}

.shortcut{
width: auto;
}

.shortcut-key{
color: $footer-key-foreground;
}
Loading