diff --git a/debugger.py b/debugger.py
new file mode 100644
index 0000000000000000000000000000000000000000..f8e711b4b24e9a87c5c9938c09be3c2af86545ce
--- /dev/null
+++ b/debugger.py
@@ -0,0 +1,3 @@
+from geomagio.apiclient.metadata import create
+
+create(station="BOU", category="reading")
\ No newline at end of file
diff --git a/geomagio/apiclient/MetadataFactory.py b/geomagio/apiclient/MetadataFactory.py
index 8ffaead94408a88c6b3aeb7591f1a66e38aa9675..1d8ded01879bd82921395f0343655b74d0bdbbe5 100644
--- a/geomagio/apiclient/MetadataFactory.py
+++ b/geomagio/apiclient/MetadataFactory.py
@@ -1,7 +1,6 @@
import os
import requests
-from typing import Dict, List, Union
-import urllib
+from typing import List, Union
from obspy import UTCDateTime
from pydantic import parse_obj_as
@@ -22,35 +21,37 @@ class MetadataFactory(object):
self.token = token
self.header = {"Authorization": self.token} if token else None
- def delete_metadata(self, metadata: Metadata) -> Dict:
+ def delete_metadata(self, metadata: Metadata) -> bool:
response = requests.delete(url=f"{self.url}/{metadata.id}", headers=self.header)
- return response
+ if response.status_code == 200:
+ return True
+ return False
- def get_metadata(self, query: MetadataQuery) -> Union[List[Metadata], Metadata]:
+ def get_metadata(self, query: MetadataQuery) -> List[Metadata]:
args = parse_params(query=query)
if "id" in args:
self.url = f"{self.url}/{args['id']}"
args = {}
- responses = requests.get(url=self.url, params=args, headers=self.header)
+ response = requests.get(url=self.url, params=args, headers=self.header)
try:
- metadata = parse_obj_as(Union[List[Metadata], Metadata], responses.json())
+ metadata = parse_obj_as(Union[List[Metadata], Metadata], response.json())
except:
- raise ValueError("Data not found")
+ return []
if isinstance(metadata, Metadata):
metadata = [metadata]
return metadata
- def create_metadata(self, metadata: Metadata) -> requests.Response:
+ def create_metadata(self, metadata: Metadata) -> Metadata:
response = requests.post(
url=self.url, data=metadata.json(), headers=self.header
)
- return response
+ return Metadata(**response.json())
- def update_metadata(self, metadata: Metadata) -> requests.Response:
+ def update_metadata(self, metadata: Metadata) -> Metadata:
response = requests.put(
url=f"{self.url}/{metadata.id}", data=metadata.json(), headers=self.header
)
- return response
+ return Metadata(**response.json())
def parse_params(query: MetadataQuery) -> str:
diff --git a/geomagio/apiclient/metadata.py b/geomagio/apiclient/metadata.py
index 42551ad4562b5fb1f901dccc5d81026229f5547c..272e2cd87a72811d0bc39746916861fc916558ec 100644
--- a/geomagio/apiclient/metadata.py
+++ b/geomagio/apiclient/metadata.py
@@ -11,18 +11,15 @@ from ..metadata import Metadata, MetadataCategory
from .MetadataFactory import MetadataFactory
-def load_metadata(input_file) -> Optional[Dict]:
+def load_metadata(input_file: str) -> Optional[Dict]:
if input_file is None:
return None
if input_file == "-":
data = json.loads(sys.stdin.read())
return data
- try:
- with open(input_file, "r") as file:
- data = json.load(file)
- return data
- except (FileNotFoundError, TypeError):
- return None
+ with open(input_file, "r") as file:
+ data = json.load(file)
+ return data
app = typer.Typer()
@@ -40,10 +37,7 @@ def create(
data_valid: bool = True,
endtime: str = None,
id: int = None,
- input_file: str = typer.Option(
- None,
- help="JSON formatted file containing non-shared metadata information",
- ),
+ input_file: str = None,
location: str = None,
metadata_valid: bool = True,
network: str = None,
@@ -51,72 +45,40 @@ def create(
station: str = None,
wrap: bool = True,
):
-
- if wrap == True:
+ input_metadata = load_metadata(input_file=input_file)
+ if not wrap:
+ metadata = Metadata(**input_metadata)
+ else:
metadata = Metadata(
- id=id,
category=category,
- starttime=UTCDateTime(starttime) if starttime else None,
- endtime=UTCDateTime(endtime) if endtime else None,
+ channel=channel,
created_after=UTCDateTime(created_after) if created_after else None,
created_before=UTCDateTime(created_before) if created_before else None,
- network=network,
- station=station,
- channel=channel,
- location=location,
data_valid=data_valid,
+ endtime=UTCDateTime(endtime) if endtime else None,
+ id=id,
+ location=location,
+ metadata = input_metadata["metadata"],
metadata_valid=metadata_valid,
+ network=network,
+ starttime=UTCDateTime(starttime) if starttime else None,
+ station=station,
)
- metadata.metadata = load_metadata(input_file=input_file)
- elif wrap == False:
- metadata_dict = load_metadata(input_file=input_file)
- metadata = Metadata(**metadata_dict)
- response = MetadataFactory(url=url).create_metadata(metadata=metadata)
- print(response.json())
+ metadata = MetadataFactory(url=url).create_metadata(metadata=metadata)
+ print(metadata.json())
@app.command()
def delete(
+ input_file: str,
url: str = "http://{}/ws/secure/metadata".format(
os.getenv("GEOMAG_API_HOST", "127.0.0.1:8000")
),
- category: MetadataCategory = None,
- channel: str = None,
- created_after: str = None,
- created_before: str = None,
- data_valid: bool = True,
- endtime: str = None,
- id: int = None,
- input_file: str = typer.Option(
- None,
- help="JSON formatted file containing non-shared metadata information",
- ),
- location: str = None,
- metadata_valid: bool = True,
- network: str = None,
- starttime: str = None,
- station: str = None,
):
- if input_file is not None:
- metadata_dict = load_metadata(input_file=input_file)
- metadata = Metadata(**metadata_dict)
- else:
- metadata = Metadata(
- id=id,
- category=category,
- starttime=starttime,
- endtime=endtime,
- created_after=created_after,
- created_before=created_before,
- network=network,
- station=station,
- channel=channel,
- location=location,
- data_valid=data_valid,
- metadata_valid=metadata_valid,
- )
- response = MetadataFactory(url=url).delete_metadata(metadata=metadata)
- print(response.json())
+ metadata_dict = load_metadata(input_file=input_file)
+ metadata = Metadata(**metadata_dict)
+ deleted = MetadataFactory(url=url).delete_metadata(metadata=metadata)
+ print(deleted)
@app.command()
@@ -136,75 +98,45 @@ def get(
network: Optional[str] = None,
starttime: Optional[str] = None,
station: Optional[str] = None,
- unwrap: bool = False,
+ getone: bool = False,
):
query = MetadataQuery(
- id=id,
category=category,
- starttime=UTCDateTime(starttime) if starttime else None,
- endtime=UTCDateTime(endtime) if endtime else None,
+ channel=channel,
created_after=UTCDateTime(created_after) if created_after else None,
created_before=UTCDateTime(created_before) if created_before else None,
- network=network,
- station=station,
- channel=channel,
- location=location,
data_valid=data_valid,
+ endtime=UTCDateTime(endtime) if endtime else None,
+ id=id,
+ location=location,
metadata_valid=metadata_valid,
+ network=network,
+ starttime=UTCDateTime(starttime) if starttime else None,
+ station=station,
)
metadata = MetadataFactory(url=url).get_metadata(query=query)
- if unwrap:
- metadata = [json.dumps(m.metadata) for m in metadata]
- else:
- metadata = [m.json() for m in metadata]
- if len(metadata) == 1:
- print(metadata[0])
+ if not metadata:
+ print([])
+ return
+
+ if getone:
+ if len(metadata) > 1:
+ raise ValueError("More than one matching record")
+ print(metadata[0].json())
return
- print(metadata)
+ print([m.json() for m in metadata])
+
@app.command()
def update(
+ input_file: str,
url: str = "http://{}/ws/secure/metadata".format(
os.getenv("GEOMAG_API_HOST", "127.0.0.1:8000")
),
- category: MetadataCategory = None,
- channel: str = None,
- created_after: str = None,
- created_before: str = None,
- data_valid: bool = True,
- endtime: str = None,
- id: int = None,
- input_file: str = typer.Option(
- None,
- help="JSON formatted file containing non-shared metadata information",
- ),
- location: str = None,
- metadata_valid: bool = True,
- network: str = None,
- starttime: str = None,
- station: str = None,
):
- if input_file is not None:
- metadata_dict = load_metadata(input_file=input_file)
- metadata = Metadata(**metadata_dict)
-
- else:
- metadata = Metadata(
- id=id,
- category=category,
- starttime=UTCDateTime(starttime) if starttime else None,
- endtime=UTCDateTime(endtime) if endtime else None,
- created_after=UTCDateTime(created_after) if created_after else None,
- created_before=UTCDateTime(created_before) if created_before else None,
- network=network,
- station=station,
- channel=channel,
- location=location,
- data_valid=data_valid,
- metadata_valid=metadata_valid,
- )
- metadata.metadata = load_metadata(input_file=input_file)
+ metadata_dict = load_metadata(input_file=input_file)
+ metadata = Metadata(**metadata_dict)
response = MetadataFactory(url=url).update_metadata(metadata=metadata)
print(response.json())