Skip to content

Commit

Permalink
Add docstrings to enhance code documentation and clarity
Browse files Browse the repository at this point in the history
  • Loading branch information
felipeadeildo committed Nov 25, 2024
1 parent bd9e1b8 commit 2a45146
Showing 1 changed file with 16 additions and 5 deletions.
21 changes: 16 additions & 5 deletions pysus/ftp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,9 @@ class FileInfo(TypedDict):

@runtime_checkable
class Downloadable(Protocol):
async def download(self, local_dir: str) -> Data: ...
async def download(self, local_dir: str) -> Data:
"""Protocol for downloadable objects"""
...


class FTPSingleton:
Expand All @@ -71,13 +73,15 @@ class FTPSingleton:

@classmethod
def get_instance(cls) -> FTP:
"""Get or create the singleton FTP instance"""
if cls._instance is None or not cls._instance.sock:
cls._instance = FTP("ftp.datasus.gov.br")
cls._instance.login()
return cls._instance

@classmethod
def close(cls) -> None:
"""Close the singleton FTP instance"""
if cls._instance and cls._instance.sock:
cls._instance.close()
cls._instance = None
Expand Down Expand Up @@ -111,8 +115,6 @@ class File:
Static method to parse a line from the FTP LIST command and extract file information.
"""

"""FTP File representation with improved type safety"""

def __init__(self, path: str, name: str, info: FileInfo) -> None:
self.name, self.extension = os.path.splitext(name)
self.basename: str = f"{self.name}{self.extension}"
Expand All @@ -126,6 +128,7 @@ def __init__(self, path: str, name: str, info: FileInfo) -> None:

@property
def info(self) -> Dict[str, str]:
"""Returns a dictionary with human-readable file information"""
return {
"size": humanize.naturalsize(self.__info["size"]),
"type": f"{self.extension[1:].upper()} file",
Expand All @@ -135,6 +138,7 @@ def info(self) -> Dict[str, str]:
def download(
self, local_dir: str = CACHEPATH, _pbar: Optional[tqdm] = None
) -> Data:
"""Downloads the file to the specified local directory"""
target_dir = pathlib.Path(local_dir)
target_dir.mkdir(exist_ok=True, parents=True)

Expand Down Expand Up @@ -178,6 +182,7 @@ def callback(data: bytes) -> None:
return Data(str(filepath), _pbar=_pbar) # type: ignore

async def async_download(self, local_dir: str = CACHEPATH) -> Data:
"""Asynchronously downloads the file to the specified local directory"""
target_dir = pathlib.Path(local_dir)
target_dir.mkdir(exist_ok=True, parents=True)
filepath = target_dir / self.basename
Expand All @@ -198,7 +203,7 @@ async def async_download(self, local_dir: str = CACHEPATH) -> Data:

@staticmethod
def _line_parser(file_line: bytes) -> Tuple[str, Dict[str, Any]]:
# Implementation moved to static method for better organization
"""Static method to parse a line from the FTP LIST command and extract file information"""
line = file_line.decode("utf-8")
if "<DIR>" in line:
date, time, _, *name = line.strip().split()
Expand Down Expand Up @@ -290,11 +295,13 @@ def __new__(cls, path: str, _is_root_child: bool = False) -> "Directory":

@staticmethod
def _normalize_path(path: str) -> str:
"""Normalizes the given path"""
path = f"/{path}" if not path.startswith("/") else path
return path[:-1] if path.endswith("/") else path
return path.removesuffix("/")

@classmethod
def _get_root_directory(cls) -> Directory:
"""Returns the root directory instance, creating it if necessary"""
if "/" not in DIRECTORY_CACHE:
root = super().__new__(cls)
root.parent = root
Expand All @@ -306,24 +313,28 @@ def _get_root_directory(cls) -> Directory:
return DIRECTORY_CACHE["/"]

def _init_root_child(self, name: str) -> None:
"""Initializes a root child directory"""
self.parent = DIRECTORY_CACHE["/"]
self.name = name
self.loaded = False
self.__content__ = {}

def _init_regular(self, parent_path: str, name: str) -> None:
"""Initializes a regular directory"""
self.parent = Directory(parent_path)
self.name = name
self.loaded = False
self.__content__ = {}

@property
def content(self) -> List[Union[Directory, File]]:
"""Returns the content of the directory, loading it if necessary"""
if not self.loaded:
self.load()
return list(self.__content__.values())

def load(self) -> Self:
"""Loads the content of the directory and marks it as loaded"""
self.__content__ |= load_directory_content(self.path)
self.loaded = True
return self
Expand Down

0 comments on commit 2a45146

Please sign in to comment.