2163816198

Build Type

github

Committed by GitHub

Commit Message

Merge 94199f2bb into 9ac032fc5

Pull Request Pull Request #24: Add CMIP file structure, use pyessv controlled vocabularies, and major refactoring

Run Details

242 of 1140 new or added lines in 35 files covered. (21.23%)

13 existing lines in 4 files now uncovered.

735 of 3278 relevant lines covered (22.42%)

0.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

15.84

/miranda/archive/remove.py

import logging
from datetime import date
from getpass import getpass
from logging import config
from pathlib import Path
from types import GeneratorType
from typing import List, Optional, Union

import fabric

from miranda.scripting import LOGGING_CONFIG
from miranda.storage import report_file_size
from miranda.utils import creation_date

config.dictConfig(LOGGING_CONFIG)


def file_emptier(*, file_list: Union[List[Union[str, Path]], GeneratorType]) -> None:
    """
    Provided a list of file paths, will open and overwrite them in order to delete data while preserving the file name.

    Parameters
    ----------
    file_list: List[Union[str, Path]]
      List of files to be overwritten

    Returns
    -------
    None
    """

    file_list = [Path(f) for f in file_list]
    file_list.sort()

    logging.info(
        f"Found {len(file_list)} files totalling {report_file_size(file_list)}."
    )

    for file in file_list:
        logging.warning(f"Overwriting {file}")
        open(file, "w").close()


def delete_by_date(
    *,
    source: Union[str, Path],
    year: Optional[int] = None,
    month: Optional[int] = None,
    day: Optional[int] = None,
    pattern: Optional[str] = None,
    server: Optional[Union[str, Path]] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
    date_object: Optional[date] = None,
) -> None:
    """

    Parameters
    ----------
    source: Union[str, Path]
    year: Optional[int]
    month: Optional[int]
    day: Optional[int]
    pattern: Optional[str]
    server: Optional[Union[str, Path]]
    user: Optional[str]
    password: Optional[str]
    date_object: Optional[date]

    Returns
    -------
    None
    """

    user = user or input("Username:")
    password = password or getpass("Password:")

    if year and month and day:
        date_selected = date(year, month, day)
    elif date_object:
        date_selected = date_object
    else:
        raise ValueError

    glob_pattern = pattern or "*.nc"

    nc_files = Path(source).rglob(glob_pattern)
    nc_files = list(nc_files)
    nc_files.sort()

    logging.info(f"Found {len(nc_files)} files totalling {report_file_size(nc_files)}.")

    context = None
    if server:
        connection = fabric.Connection(
            host=server, user=user, connect_kwargs=dict(password=password)
        )
        context = connection.open()

    freed_space = 0
    deleted_files = 0

    for file in nc_files:
        if creation_date(file) == date_selected:
            freed_space += Path(file).stat().st_size
            logging.info(f"Deleting {file.name}")
            if context:
                context.remove(file)
            else:
                file.unlink()
            deleted_files += 1

    logging.info(
        f"Removed {deleted_files} files totalling {report_file_size(freed_space)}"
    )

    if server:
        context.close()

    return


def delete_duplicates(
    *,
    source: Union[str, Path],
    target: Union[str, Path],
    server: Optional[Union[str, Path]],
    user: str = None,
    password: str = None,
    pattern: str = None,
    delete_target_duplicates: bool = False,
) -> None:
    """

    Parameters
    ----------
    source : Union[str, Path]
    target : Union[str, Path]
    server : Optional[Union[str, Path]]
    user: str
    password : str
    pattern: str
    delete_target_duplicates : bool

    Returns
    -------
    None
    """

    user = user or input("Username:")
    password = password or getpass("Password:")
    glob_pattern = pattern or "*.nc"

    connection = fabric.Connection(
        host=server, user=user, connect_kwargs=dict(password=password)
    )

    nc_files_source = Path(source).rglob(glob_pattern)
    nc_files_source = {f.stem for f in nc_files_source}
    nc_files_target = Path(target).rglob(glob_pattern)

    nc_file_duplicates = []
    for f in nc_files_target:
        if f.name in nc_files_source:
            logging.info(f"Duplicate found: {f.name}")
            nc_file_duplicates.append(f)

    nc_file_duplicates.sort()
    logging.info(
        f"Found {len(nc_file_duplicates)} files totalling {report_file_size(nc_file_duplicates)}"
    )

    freed_space = 0
    deleted_files = 0
    if delete_target_duplicates:
        with connection as context:
            for dup in nc_file_duplicates:
                freed_space += Path(dup).stat().st_size
                logging.info(f"Deleting {dup.name}")
                context.remove(dup)
                deleted_files += 1

    logging.info(
        f"Removed { deleted_files} files totalling {report_file_size(freed_space)}."
    )
    return


def delete_by_variable(
    *,
    target: Union[str, Path, List[Union[str, Path]], GeneratorType] = None,
    variables: List[str] = None,
    server: Optional[str or Path],
    user: Optional[str] = None,
    password: Optional[str] = None,
    file_suffix: Optional[str] = None,
    delete: bool = False,
) -> None:
    """
    Given target location(s), a list of variables and a server address, perform a glob search
     and delete file names starting with the variables identified

    Parameters
    ----------
    target : Union[str, Path, List[Union[str, Path]], GeneratorType]
    variables : List[str]
    server :Optional[Union[str, Path]]
    user : Optional[str]
    password : Optional[str]
    file_suffix : Optional[str]
    delete : bool

    Returns
    -------
    None
    """

    user = user or input("Username:")
    password = password or getpass("Password:")

    connection = fabric.Connection(
        host=server, user=user, connect_kwargs=dict(password=password)
    )

    freed_space = 0
    deleted_files = 0
    for var in variables:
        glob_suffix = file_suffix or ".nc"

        if isinstance(target, (GeneratorType, list)):
            found = list()
            for location in target:
                found.extend([f for f in Path(location).rglob(f"{var}*{glob_suffix}")])
        else:
            found = Path(target).rglob(f"{var}*{glob_suffix}")

        nc_files = [Path(f) for f in found]
        nc_files.sort()

        logging.info(
            f"Found {len(nc_files)} files totalling {report_file_size(nc_files)}"
        )

        with connection as context:
            for file in nc_files:
                freed_space += Path(file).stat().st_size
                deleted_files += 1
                if delete:
                    logging.info(f"Deleting file {file.stem}")
                    context.remove(file)

    logging.info(
        f"Removed {deleted_files} files totalling {report_file_size(freed_space)}"
    )
    return

1	import logging	3✔
2	from datetime import date	3✔
3	from getpass import getpass	3✔
4	from logging import config	3✔
5	from pathlib import Path	3✔
6	from types import GeneratorType	3✔
7	from typing import List, Optional, Union	3✔
8
9	import fabric	3✔
10
11	from miranda.scripting import LOGGING_CONFIG	3✔
12	from miranda.storage import report_file_size	3✔
13	from miranda.utils import creation_date	3✔
14
15	config.dictConfig(LOGGING_CONFIG)	3✔
16
17
18	def file_emptier(*, file_list: Union[List[Union[str, Path]], GeneratorType]) -> None:	3✔
19	"""
20	Provided a list of file paths, will open and overwrite them in order to delete data while preserving the file name.
21
22	Parameters
23	----------
24	file_list: List[Union[str, Path]]
25	List of files to be overwritten
26
27	Returns
28	-------
29	None
30	"""
31
NEW 32	file_list = [Path(f) for f in file_list]	×
NEW 33	file_list.sort()	×
34
35	logging.info(	×
36	f"Found {len(file_list)} files totalling {report_file_size(file_list)}."
37	)
38
39	for file in file_list:	×
NEW 40	logging.warning(f"Overwriting {file}")	×
41	open(file, "w").close()	×
42
43
44	def delete_by_date(	3✔
45	*,
46	source: Union[str, Path],
47	year: Optional[int] = None,
48	month: Optional[int] = None,
49	day: Optional[int] = None,
50	pattern: Optional[str] = None,
51	server: Optional[Union[str, Path]] = None,
52	user: Optional[str] = None,
53	password: Optional[str] = None,
54	date_object: Optional[date] = None,
55	) -> None:
56	"""
57
58	Parameters
59	----------
60	source: Union[str, Path]
61	year: Optional[int]
62	month: Optional[int]
63	day: Optional[int]
64	pattern: Optional[str]
65	server: Optional[Union[str, Path]]
66	user: Optional[str]
67	password: Optional[str]
68	date_object: Optional[date]
69
70	Returns
71	-------
72	None
73	"""
74
75	user = user or input("Username:")	×
76	password = password or getpass("Password:")	×
77
78	if year and month and day:	×
79	date_selected = date(year, month, day)	×
80	elif date_object:	×
81	date_selected = date_object	×
82	else:
83	raise ValueError	×
84
85	glob_pattern = pattern or "*.nc"	×
86
87	nc_files = Path(source).rglob(glob_pattern)	×
88	nc_files = list(nc_files)	×
89	nc_files.sort()	×
90
NEW 91	logging.info(f"Found {len(nc_files)} files totalling {report_file_size(nc_files)}.")	×
92
93	context = None	×
94	if server:	×
95	connection = fabric.Connection(	×
96	host=server, user=user, connect_kwargs=dict(password=password)
97	)
98	context = connection.open()	×
99
100	freed_space = 0	×
101	deleted_files = 0	×
102
103	for file in nc_files:	×
104	if creation_date(file) == date_selected:	×
105	freed_space += Path(file).stat().st_size	×
NEW 106	logging.info(f"Deleting {file.name}")	×
107	if context:	×
108	context.remove(file)	×
109	else:
110	file.unlink()	×
111	deleted_files += 1	×
112
113	logging.info(	×
114	f"Removed {deleted_files} files totalling {report_file_size(freed_space)}"
115	)
116
117	if server:	×
118	context.close()	×
119
120	return	×
121
122
123	def delete_duplicates(	3✔
124	*,
125	source: Union[str, Path],
126	target: Union[str, Path],
127	server: Optional[Union[str, Path]],
128	user: str = None,
129	password: str = None,
130	pattern: str = None,
131	delete_target_duplicates: bool = False,
132	) -> None:
133	"""
134
135	Parameters
136	----------
137	source : Union[str, Path]
138	target : Union[str, Path]
139	server : Optional[Union[str, Path]]
140	user: str
141	password : str
142	pattern: str
143	delete_target_duplicates : bool
144
145	Returns
146	-------
147	None
148	"""
149
150	user = user or input("Username:")	×
151	password = password or getpass("Password:")	×
152	glob_pattern = pattern or "*.nc"	×
153
154	connection = fabric.Connection(	×
155	host=server, user=user, connect_kwargs=dict(password=password)
156	)
157
158	nc_files_source = Path(source).rglob(glob_pattern)	×
159	nc_files_source = {f.stem for f in nc_files_source}	×
160	nc_files_target = Path(target).rglob(glob_pattern)	×
161
162	nc_file_duplicates = []	×
163	for f in nc_files_target:	×
164	if f.name in nc_files_source:	×
NEW 165	logging.info(f"Duplicate found: {f.name}")	×
166	nc_file_duplicates.append(f)	×
167
168	nc_file_duplicates.sort()	×
169	logging.info(	×
170	f"Found {len(nc_file_duplicates)} files totalling {report_file_size(nc_file_duplicates)}"
171	)
172
173	freed_space = 0	×
174	deleted_files = 0	×
175	if delete_target_duplicates:	×
176	with connection as context:	×
177	for dup in nc_file_duplicates:	×
178	freed_space += Path(dup).stat().st_size	×
NEW 179	logging.info(f"Deleting {dup.name}")	×
180	context.remove(dup)	×
181	deleted_files += 1	×
182
183	logging.info(	×
184	f"Removed { deleted_files} files totalling {report_file_size(freed_space)}."
185	)
186	return	×
187
188
189	def delete_by_variable(	3✔
190	*,
191	target: Union[str, Path, List[Union[str, Path]], GeneratorType] = None,
192	variables: List[str] = None,
193	server: Optional[str or Path],
194	user: Optional[str] = None,
195	password: Optional[str] = None,
196	file_suffix: Optional[str] = None,
197	delete: bool = False,
198	) -> None:
199	"""
200	Given target location(s), a list of variables and a server address, perform a glob search
201	and delete file names starting with the variables identified
202
203	Parameters
204	----------
205	target : Union[str, Path, List[Union[str, Path]], GeneratorType]
206	variables : List[str]
207	server :Optional[Union[str, Path]]
208	user : Optional[str]
209	password : Optional[str]
210	file_suffix : Optional[str]
211	delete : bool
212
213	Returns
214	-------
215	None
216	"""
217
218	user = user or input("Username:")	×
219	password = password or getpass("Password:")	×
220
221	connection = fabric.Connection(	×
222	host=server, user=user, connect_kwargs=dict(password=password)
223	)
224
225	freed_space = 0	×
226	deleted_files = 0	×
227	for var in variables:	×
228	glob_suffix = file_suffix or ".nc"	×
229
230	if isinstance(target, (GeneratorType, list)):	×
231	found = list()	×
232	for location in target:	×
NEW 233	found.extend([f for f in Path(location).rglob(f"{var}*{glob_suffix}")])	×
234	else:
NEW 235	found = Path(target).rglob(f"{var}*{glob_suffix}")	×
236
237	nc_files = [Path(f) for f in found]	×
238	nc_files.sort()	×
239
240	logging.info(	×
241	f"Found {len(nc_files)} files totalling {report_file_size(nc_files)}"
242	)
243
244	with connection as context:	×
245	for file in nc_files:	×
246	freed_space += Path(file).stat().st_size	×
247	deleted_files += 1	×
248	if delete:	×
NEW 249	logging.info(f"Deleting file {file.stem}")	×
250	context.remove(file)	×
251
252	logging.info(	×
253	f"Removed {deleted_files} files totalling {report_file_size(freed_space)}"
254	)
255	return	×

Ouranosinc / miranda / 2163816198

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous