mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
import datetime
|
|
import pytz
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
from backend.app.db.models.solar import Solar
|
|
|
|
|
|
def get_solar_data(session: Session, longitude: float = None, latitude: float = None, uprn: str = None):
|
|
"""
|
|
This function will fetch data from the solar table based on longitude and latitude or UPRN.
|
|
:param session: The database session
|
|
:param longitude: The longitude to search for
|
|
:param latitude: The latitude to search for
|
|
:param uprn: The UPRN to search for (overrides longitude and latitude if provided)
|
|
:return: The google_api_response and updated_at fields
|
|
"""
|
|
try:
|
|
if uprn:
|
|
# Search by UPRN
|
|
solar_data = session.query(Solar.google_api_response, Solar.updated_at).filter_by(uprn=uprn).one()
|
|
else:
|
|
# Search by longitude and latitude
|
|
solar_data = session.query(Solar.google_api_response, Solar.updated_at).filter(
|
|
Solar.longitude == longitude,
|
|
Solar.latitude == latitude
|
|
).one()
|
|
|
|
# Check if updated_at is more than 6 months old
|
|
six_months_ago = datetime.datetime.now(pytz.utc) - datetime.timedelta(days=6 * 30) # Approximate 6 months
|
|
is_outdated = solar_data.updated_at < six_months_ago
|
|
|
|
return solar_data.google_api_response, solar_data.updated_at, is_outdated
|
|
|
|
except NoResultFound:
|
|
return None, None, False
|
|
|
|
|
|
def store_batch_data(session: Session, api_data: dict, uprns_to_location: list):
|
|
"""
|
|
This function will store the API data to the solar table against all of the UPRNs with longitude and latitude.
|
|
:param session: The database session
|
|
:param api_data: The API data to store
|
|
:param data_list: A list of dictionaries containing uprn, longitude, and latitude
|
|
"""
|
|
try:
|
|
# Convert the data_list to a list of dicts for bulk insert
|
|
records_to_update = []
|
|
for data in uprns_to_location:
|
|
record = {
|
|
'uprn': data['uprn'],
|
|
'longitude': data['longitude'],
|
|
'latitude': data['latitude'],
|
|
'google_api_response': api_data,
|
|
'updated_at': datetime.datetime.now(pytz.utc)
|
|
}
|
|
records_to_update.append(record)
|
|
|
|
# Perform bulk insert or update
|
|
session.bulk_insert_mappings(Solar, records_to_update)
|
|
session.commit()
|
|
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise e
|