Fix flake8

Author: webb-ben

src/wqie/__init__.py

@@ -14,4 +14,4 @@
 defs = Definitions(
     assets=[county_stations],
     jobs=[process_county_stations],
-)
+)

src/wqie/models.py

@@ -11,6 +11,7 @@
 from pydantic import BaseModel
 from typing import List, Dict, Any
 
+
 class Station(BaseModel):
     station_id: str
     name: str
@@ -23,5 +24,6 @@ class Station(BaseModel):
     provider: str
     datastreams: List[Dict[str, Any]]
 
+
 class StationsData(BaseModel):
-    stations: List[Station]
+    stations: List[Station]

src/wqie/ops/datastreams.py

@@ -1,19 +1,20 @@
-import logging
+
 from csv import DictReader
 from datetime import datetime
 from io import StringIO
 from pytz import timezone
 import re
 from requests import Session
-from typing import Iterable, Dict, List
-from dagster import op, job, Out, get_dagster_logger
+from typing import Iterable
+from dagster import get_dagster_logger
 
 from wqie.env import RESULTS_URL, NLDI_URL
 from wqie.util import make_uuid, url_join
 from wqie.mapping import MAPPING
 
 LOGGER = get_dagster_logger()
 
+
 def fetch_datastreams(station_id: str):
     """
     Load datasets from USBR RISE API
@@ -59,7 +60,10 @@ def yield_datastreams(station_identifier: str,
         kwargs = {}
         monitoring_location_identifier = \
             dataset['MonitoringLocationIdentifier']
-        url = url_join(NLDI_URL, "linked-data/wqp", monitoring_location_identifier)
+        url = url_join(
+            NLDI_URL,
+            "linked-data/wqp",
+            monitoring_location_identifier)
         try:
             result = http.get(url)
             feature = result.json()['features'][0]

src/wqie/ops/fetch.py

@@ -9,7 +9,6 @@
 # =================================================================
 
 from dagster import op, Out
-from io import BytesIO
 import requests
 from typing import Dict, Any, List
 import csv
@@ -18,7 +17,7 @@
 
 
 @op(out={"sites": Out(list)})
-def fetch_station_metadata( county: str) -> List[Dict[str, Any]]:
+def fetch_station_metadata(county: str) -> List[Dict[str, Any]]:
     """
     Fetch station metadata for a single county.
     """
@@ -30,12 +29,13 @@ def fetch_station_metadata( county: str) -> List[Dict[str, Any]]:
     }
     response = requests.get(RESULTS_URL, params=params)
     response.raise_for_status()
-    
+
     stations = []
     for row in csv.DictReader(response.text.splitlines()):
         stations.append(row)
     return stations
 
+
 @op(out={"station_details": Out(list)})
 def fetch_site_metadata(site_ids: List[str]) -> List[Dict[str, Any]]:
     """
@@ -47,7 +47,7 @@ def fetch_site_metadata(site_ids: List[str]) -> List[Dict[str, Any]]:
     }
     response = requests.post(STATION_URL, data=params)
     response.raise_for_status()
-    
+
     details = []
     for row in csv.DictReader(response.text.splitlines()):
         details.append(row)

src/wqie/ops/transform.py

@@ -8,7 +8,6 @@
 #
 # =================================================================
 
-import logging
 from dagster import op, get_dagster_logger
 from typing import List, Dict, Any
 import requests
@@ -29,13 +28,14 @@ def upsert_collection_item(collection: str, item: Dict[str, Any]) -> bool:
     response = requests.post(api_url, json=item)
     return response.status_code == 201  # Return True if creation is successful
 
+
 @op
 def transform_stations(stations: List[Dict[str, Any]]) -> StationsData:
     """
     Transform raw station data into structured format.
     """
     transformed_stations = []
-    
+
     for row in stations:
         station = Station(
             station_id=row['MonitoringLocationIdentifier'],
@@ -50,27 +50,28 @@ def transform_stations(stations: List[Dict[str, Any]]) -> StationsData:
             datastreams=[]
         )
         transformed_stations.append(station)
-    
+
     return StationsData(stations=transformed_stations)
 
+
 @op
 def publish_station_collection(stations_data: StationsData) -> None:
     """
     Publishes station collection to API config and backend.
-    
+
     Takes in the transformed station data and sends it to the SensorThings API.
-    
+
     :param stations_data: The transformed station data.
     :returns: `None`
     """
     # Iterate over the transformed stations data
     for station in stations_data.stations:
         station_identifier = station.station_id
-        
+
         try:
             datastreams = load_datastreams(station_identifier)
         except Exception:
-            LOGGER.warning(f"Failed to load datastreams for {station_identifier}")
+            LOGGER.warning(f"Failed to load datastreams for {station_identifier}") # noqa
             continue
 
         try:
@@ -100,9 +101,9 @@ def publish_station_collection(stations_data: StationsData) -> None:
             }],
             'properties': {
                 'mainstem': mainstem,
-                'hu08': url_join(GEOCONNEX_URL, 'ref/hu08', station.huc_code),
-                'state': url_join(GEOCONNEX_URL, 'ref/states', station.state_code),
-                'county': url_join(GEOCONNEX_URL, 'ref/counties', f"{station.state_code}{station.county_code}"),
+                'hu08': url_join(GEOCONNEX_URL, 'ref/hu08', station.huc_code), # noqa
+                'state': url_join(GEOCONNEX_URL, 'ref/states', station.state_code), # noqa
+                'county': url_join(GEOCONNEX_URL, 'ref/counties', f"{station.state_code}{station.county_code}"), # noqa
                 'provider': station.provider
             },
             'Datastreams': list(datastreams)
@@ -118,6 +119,7 @@ def publish_station_collection(stations_data: StationsData) -> None:
 
     return
 
+
 def get_mainstem_uri(id):
     # Convert the input geom to GeoJSON using Shapely
 

src/wqie/partitions.py

@@ -11,9 +11,9 @@
 from dagster import StaticPartitionsDefinition, get_dagster_logger
 import requests
 from typing import List
-import logging
 
-logger = logging.getLogger(__name__)
+LOGGER = get_dagster_logger()
+
 
 def load_us_counties() -> List[str]:
     """
@@ -22,24 +22,25 @@ def load_us_counties() -> List[str]:
     url = "https://reference.geoconnex.us/collections/counties/items"
     params = {"limit": 100}
     counties = []
-    get_dagster_logger().debug(url)
+    LOGGER.debug(url)
     while True:
         response = requests.get(url, params=params)
         response.raise_for_status()
         data = response.json()
-        
+
         for feature in data["features"]:
             props = feature["properties"]
             county_code = f"US:{props['statefp']}:{props['countyfp']}"
             counties.append(county_code)
-            
+
         if len(data["features"]) < params["limit"]:
             break
-            
+
         params["offset"] = params.get("offset", 0) + params["limit"]
-    
+
     return sorted(counties)
 
+
 county_partitions = StaticPartitionsDefinition(
    load_us_counties()
 )

src/wqie/util.py

@@ -4,6 +4,7 @@
 from typing import Union, Any
 from uuid import UUID
 
+
 def get_typed_value(value) -> Union[float, int, str]:
     """
     Derive true type from data value
@@ -25,6 +26,7 @@ def get_typed_value(value) -> Union[float, int, str]:
 
     return value2
 
+
 def extract_coord(p):
     """
     helper function to extract coordinate
@@ -46,6 +48,7 @@ def clean_word(input: str, delim: str = ' ') -> str:
     """
     return delim.join(re.findall(r'\w+', input))
 
+
 def url_join(*parts: str) -> str:
     """
     helper function to join a URL from a number of parts/fragments.
@@ -61,6 +64,7 @@ def url_join(*parts: str) -> str:
 
     return '/'.join([str(p).strip().strip('/') for p in parts]).rstrip('/')
 
+
 def get_env(key: str, fallback: Any = None) -> str:
     """Fetch environment variable"""
     val = os.environ.get(key, fallback)
@@ -69,6 +73,7 @@ def get_env(key: str, fallback: Any = None) -> str:
 
     return val
 
+
 def make_uuid(input: str, raw: bool = False) -> UUID:
     """
     helper function to make uuid