5115668225

pending completion

Build # 5115668225

Build Type

push

github

Committed by

web-flow

Commit Message

Fix memoization (#105)

Run Details

117 of 127 branches covered (92.13%)

Branch coverage included in aggregate %.

86 of 87 new or added lines in 8 files covered. (98.85%)

2 existing lines in 1 file now uncovered.

479 of 490 relevant lines covered (97.76%)

1.96 hits per line

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

94.12

/src/stocktracer/cache.py

"""This module takes care of managing caching configuration."""
import atexit
import hashlib
import os
from datetime import timedelta
from pathlib import Path

from beartype import beartype
from diskcache import Cache
from platformdirs import user_cache_dir
from requests_cache import CachedSession, SQLiteCache


@beartype
def get_cache_dir() -> Path:
    """Get the cache directory used by stocktracer.

    Users can customize this directory on all systems using `STOCKTRACER_CACHE_DIR`
    environment variable. By default, the cache directory is the user cache directory
    under the stocktracer application.

    This result is immediately set to a constant `stocktracer.cache.CACHE_DIR` as to avoid
    repeated calls.

    Returns:
        Path: path to the cache directory
    """
    default_cache_dir = user_cache_dir("stocktracer")
    cache_dir = Path(os.environ.get("STOCKTRACER_CACHE_DIR", default_cache_dir))
    return cache_dir


CACHE_DIR = get_cache_dir()
CACHE_DIR.mkdir(parents=True, exist_ok=True)

results = Cache(directory=CACHE_DIR / "results", tag_index=True)


sec_data = CachedSession(
    "data",
    backend=SQLiteCache(db_path=CACHE_DIR / "data"),
    serializer="pickle",
    expire_after=timedelta(days=365 * 5),
    stale_if_error=True,
)
sec_tickers = CachedSession(
    "tickers",
    backend=SQLiteCache(db_path=CACHE_DIR / "tickers"),
    expire_after=timedelta(days=365),
    stale_if_error=True,
)


dir_name = os.path.dirname(__file__)
with open(os.path.join(dir_name, "collector/sec.py"), encoding="utf8") as opened_file:
    readFile = opened_file.read()


SEC_FILE_HASH = hashlib.sha256(readFile.encode()).hexdigest()

if SEC_FILE_HASH != results.get("sec_file_hash"):
    results.evict(tag="sec")
    results.evict(tag="results")


def update_modified_file_cache() -> None:
    """Update the cache that keeps track of when files are modified."""
    results.set("sec_file_hash", SEC_FILE_HASH)
    results.close()


atexit.register(update_modified_file_cache)

1	"""This module takes care of managing caching configuration."""
2	import atexit	2✔
3	import hashlib	2✔
4	import os	2✔
5	from datetime import timedelta	2✔
6	from pathlib import Path	2✔
7
8	from beartype import beartype	2✔
9	from diskcache import Cache	2✔
10	from platformdirs import user_cache_dir	2✔
11	from requests_cache import CachedSession, SQLiteCache	2✔
12
13
14	@beartype	2✔
15	def get_cache_dir() -> Path:	2✔
16	"""Get the cache directory used by stocktracer.
17
18	Users can customize this directory on all systems using `STOCKTRACER_CACHE_DIR`
19	environment variable. By default, the cache directory is the user cache directory
20	under the stocktracer application.
21
22	This result is immediately set to a constant `stocktracer.cache.CACHE_DIR` as to avoid
23	repeated calls.
24
25	Returns:
26	Path: path to the cache directory
27	"""
28	default_cache_dir = user_cache_dir("stocktracer")	2✔
29	cache_dir = Path(os.environ.get("STOCKTRACER_CACHE_DIR", default_cache_dir))	2✔
30	return cache_dir	2✔
31
32
33	CACHE_DIR = get_cache_dir()	2✔
34	CACHE_DIR.mkdir(parents=True, exist_ok=True)	2✔
35
36	results = Cache(directory=CACHE_DIR / "results", tag_index=True)	2✔
37
38
39	sec_data = CachedSession(	2✔
40	"data",
41	backend=SQLiteCache(db_path=CACHE_DIR / "data"),
42	serializer="pickle",
43	expire_after=timedelta(days=365 * 5),
44	stale_if_error=True,
45	)
46	sec_tickers = CachedSession(	2✔
47	"tickers",
48	backend=SQLiteCache(db_path=CACHE_DIR / "tickers"),
49	expire_after=timedelta(days=365),
50	stale_if_error=True,
51	)
52
53
54	dir_name = os.path.dirname(__file__)	2✔
55	with open(os.path.join(dir_name, "collector/sec.py"), encoding="utf8") as opened_file:	2✔
56	readFile = opened_file.read()	2✔
57
58
59	SEC_FILE_HASH = hashlib.sha256(readFile.encode()).hexdigest()	2✔
60
61	if SEC_FILE_HASH != results.get("sec_file_hash"):	2✔
62	results.evict(tag="sec")	2✔
63	results.evict(tag="results")	2✔
64
65
66	def update_modified_file_cache() -> None:	2✔
67	"""Update the cache that keeps track of when files are modified."""
68	results.set("sec_file_hash", SEC_FILE_HASH)	×
NEW 69	results.close()	×
70
71
72	atexit.register(update_modified_file_cache)	2✔

gyund / fundamental-analysis / 5115668225

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