|“`
import typing
import os
from datetime import datetime, timedelta
from pydantic import BaseModel
Image: www.cashbackforex.com
class QueryIntervalModel(BaseModel):
from_date: datetime
to_date: datetime
user_id: typing.Optional[int] = None
def generate_query_by_daterange(data: QueryIntervalModel, payload: dict) -> dict:
“””Generate query dict to retrieve data from database within directory if within the specified date range”””
from_date = data.from_date.replace(hour=0, minute=0, second=0, microsecond=0)
to_date = data.to_date.replace(hour=23, minute=59, second=59, microsecond=999999)
key = str(fromdate) + “” + str(todate) + “” + str(data.user_id)
query =
"query":
"bool":
"must": [
"match_phrase": "from": data.from_date.isoformat()
,
"match_phrase": "from": data.to_date.isoformat()
,
]
if data.user_id:
if os.path.exists(os.path.join(os.getcwd(), "data", str(key))):
query =
"query":
"bool":
"must": []
with open(
os.path.join(os.getcwd(), "data", key), "r"
) as dump_file:
count = 0
for line in dump_file:
line = line[:-1]
data = payload.copy()
data["interacted_with"] = []
data["interacted_with"].append(line)
if count == 0:
data["transmission_path"] = []
query["query"]["bool"]["must"].append(
"match_phrase":
"user_id": line
)
else:
query["query"]["bool"]["must"].append(
"nested":
"path": "transmission_path",
"query":
"bool":
"must": [
"match_phrase":
"transmission_path.user_id": line
]
)
count += 1
return query
else:
query =
"query":
"bool":
"must": [
"match_phrase": "user_id": data.user_id
,
"range":
"from": from_date.isoformat(),
"to": to_date.isoformat(),
"include_lower": True,
"include_upper": True,
]
return query
return querydef generate_query(data: datetime, payload: dict) -> dict:
“””Query dict for querying data for specific date”””
return
“query”:
“bool”:
“must”: [
“match_phrase”: “date”: data.isoformat()
]
Image: www.tradeforexnigeria.com
Forex Trading Hours In South Africa Time
https://youtube.com/watch?v=rqA0Mbtj20I
