feature: Added all the IPNS APIs and tested

Author: unniznd

src/lighthouseweb3/__init__.py

@@ -2,7 +2,16 @@
 
 import os
 import io
-from .functions import upload as d,deal_status, get_uploads as getUploads, download as _download
+from .functions import (
+    upload as d,
+    deal_status, 
+    get_uploads as getUploads, 
+    download as _download,
+    ipns_generate_key as ipnsGenerateKey,
+    ipns_publish_record as ipnsPublishRecord,
+    get_ipns_record as getIpnsRecord,
+    remove_ipns_record as removeIpnsRecord,
+)
 
 
 class Lighthouse:
@@ -38,6 +47,54 @@ def uploadBlob(self, source: io.BufferedReader, filename: str, tag: str = ''):
             return d.uploadBlob(source, filename, self.token, tag)
         except Exception as e:
             raise e
+    
+    def generateKey(self):
+        """
+        Generate a new IPNS key for the authenticated user.
+        :return: dict, The generated IPNS key information.
+        """
+        try:
+            return ipnsGenerateKey.ipns_generate_key(self.token)
+        except Exception as e:
+            raise e
+    
+    def publishRecord(self, cid: str, keyName: str):
+        """
+        Publish an IPNS record for a given CID and key name.
+
+        :param cid: str, Content Identifier to publish
+        :param keyName: str, Name of the IPNS key to use
+        :return: dict, The published IPNS record information
+        """
+        try:
+            return ipnsPublishRecord.ipns_publish_record(self.token, cid, keyName)
+        except Exception as e:
+            raise e
+    
+    def getAllKeys(self):
+        """
+        Retrieves all IPNS records associated with the current token.
+
+        return: list A list of IPNS records retrieved using the provided token.
+        """
+
+        try:
+            return getIpnsRecord.get_ipns_records(self.token)
+        except Exception as e:
+            raise e
+    
+    def removeKey(self, keyName: str):
+        """
+        Remove IPNS record of the given keyName
+
+        :param keyName: str, Name of the IPNS key to use
+        :return: dict, A dict of removed IPNS record.
+        """
+
+        try:
+            return removeIpnsRecord.remove_ipns_record(self.token, keyName)
+        except Exception as e:
+            raise e
 
     @staticmethod
     def downloadBlob(dist: io.BufferedWriter, cid: str, chunk_size=1024*1024*10):

src/lighthouseweb3/functions/get_ipns_record.py

@@ -0,0 +1,19 @@
+from .config import Config
+import requests as req
+
+def get_ipns_records(token: str):
+  headers = {
+    "Authorization": f"Bearer {token}",
+  }
+  url = f"{Config.lighthouse_api}/api/ipns/get_ipns_records"
+  try:
+    response = req.get(url, headers=headers)
+  except Exception as e:
+    raise Exception("Failed to get ipns records")
+
+  if response.status_code != 200:
+    return response.json()
+
+  return {
+    "data": response.json()
+  }

src/lighthouseweb3/functions/ipns_generate_key.py

@@ -0,0 +1,19 @@
+from .config import Config
+import requests as req
+
+def ipns_generate_key(token: str):
+  headers = {
+    "Authorization": f"Bearer {token}",
+  }
+  url = f"{Config.lighthouse_api}/api/ipns/generate_key"
+  try:
+    response = req.get(url, headers=headers)
+  except Exception as e:
+    raise Exception("Failed to ipns generate key")
+
+  if response.status_code != 200:
+    return response.json()
+
+  return {
+    "data": response.json()
+  }

src/lighthouseweb3/functions/ipns_publish_record.py

@@ -0,0 +1,19 @@
+from .config import Config
+import requests as req
+
+def ipns_publish_record(token: str, cid: str, keyName: str):
+  headers = {
+    "Authorization": f"Bearer {token}",
+  }
+  url = f"{Config.lighthouse_api}/api/ipns/publish_record?cid={cid}&keyName={keyName}"
+  try:
+    response = req.get(url, headers=headers)
+  except Exception as e:
+    raise Exception("Failed to ipns generate key")
+
+  if response.status_code != 200:
+    return response.json()
+
+  return {
+    "data": response.json()
+  }

src/lighthouseweb3/functions/remove_ipns_record.py

@@ -0,0 +1,19 @@
+from .config import Config
+import requests as req
+
+def remove_ipns_record(token: str,  keyName: str):
+  headers = {
+    "Authorization": f"Bearer {token}",
+  }
+  url = f"{Config.lighthouse_api}/api/ipns/remove_key?keyName={keyName}"
+  try:
+    response = req.delete(url, headers=headers)
+  except Exception as e:
+    raise Exception("Failed to remove ipns record")
+
+  if response.status_code != 200:
+    return response.json()
+
+  return {
+    "data": response.json()
+  }

tests/test_get_ipns_record.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+import os
+import unittest
+from src.lighthouseweb3 import Lighthouse
+from .setup import parse_env
+
+
+class TestGetIPNSRecord(unittest.TestCase):
+
+    def test_get_ipns_records(self):
+        """test get_ipns_records function"""
+        parse_env()
+        l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
+        res = l.getAllKeys()
+        self.assertIsInstance(res, dict, "result is a dict")
+        self.assertIsInstance(res.get("data"), list , "data is a dict")
+        self.assertIsInstance(res.get("data")[0].get("ipnsName"), str , "ipnsName is a str")
+        self.assertIsInstance(res.get("data")[0].get("ipnsId"), str , "ipnsId is a str")
+        self.assertIsInstance(res.get("data")[0].get("publicKey"), str , "publicKey is a str")
+        self.assertIsInstance(res.get("data")[0].get("cid"), str , "cid is a str")
+        self.assertIsInstance(res.get("data")[0].get("lastUpdate"), int , "lastUpdate is a int")

tests/test_ipns_generate_key.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+import os
+import unittest
+from src.lighthouseweb3 import Lighthouse
+from .setup import parse_env
+
+
+class TestIPNSGenerateKey(unittest.TestCase):
+
+    def test_ipns_generate_key(self):
+        """test ipns_generate_key function"""
+        parse_env()
+        l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
+        res = l.generateKey()
+        self.assertIsInstance(res, dict, "result is a dict")
+        self.assertIsInstance(res.get("data"), dict , "data is a dict")
+        self.assertIsInstance(res.get("data").get("ipnsName"), str , "ipnsName is a str")
+        self.assertIsInstance(res.get("data").get("ipnsId"), str , "ipnsId is a dict")

tests/test_ipns_publish_record.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+import os
+import unittest
+from src.lighthouseweb3 import Lighthouse
+from .setup import parse_env
+
+
+class TestIPNSPublishRecord(unittest.TestCase):
+
+    def test_ipns_publish_record(self):
+        """test ipns_publish_record function"""
+        parse_env()
+
+        l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
+        res = l.generateKey()
+
+        self.assertIsInstance(res, dict, "result is a dict")
+        self.assertIsInstance(res.get("data"), dict , "data is a dict")
+        self.assertIsInstance(res.get("data").get("ipnsName"), str , "ipnsName is a str")
+        self.assertIsInstance(res.get("data").get("ipnsId"), str , "ipnsId is a dict")
+
+        record = l.publishRecord(
+          'QmeMsykMDyD76zpAbinCy1cjb1KL6CVNBfB44am15U1XHh', 
+          res.get('data').get('ipnsName')
+        )
+
+        self.assertIsInstance(record, dict, "record is a dict")
+        self.assertIsInstance(record.get("data"), dict, "data is a dict")
+        self.assertIsInstance(record.get("data").get("Name"), str, "name is a str")
+        self.assertIsInstance(record.get("data").get("Value"), str, "value is a str")
+        self.assertEqual(record.get("data").get("Value"), "/ipfs/QmeMsykMDyD76zpAbinCy1cjb1KL6CVNBfB44am15U1XHh")

tests/test_remove_ipns_record.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+import os
+import unittest
+from src.lighthouseweb3 import Lighthouse
+from .setup import parse_env
+
+
+class TestRemoveIPNSRecord(unittest.TestCase):
+
+    def test_ipns_generate_key(self):
+        """test ipns_generate_key function"""
+        parse_env()
+        l = Lighthouse(os.environ.get("LIGHTHOUSE_TOKEN"))
+
+        key = l.generateKey()
+
+        self.assertIsInstance(key, dict, "result is a dict")
+        self.assertIsInstance(key.get("data"), dict , "data is a dict")
+        self.assertIsInstance(key.get("data").get("ipnsName"), str , "ipnsName is a str")
+        self.assertIsInstance(key.get("data").get("ipnsId"), str , "ipnsId is a dict")
+
+    
+        res = l.removeKey(key.get('data').get('ipnsName'))
+        self.assertIsInstance(res, dict, "result is a dict")
+        self.assertIsInstance(res.get("data"), dict , "data is a dict")
+        self.assertIsInstance(res.get("data").get("Keys"), list , "Keys is a list")
+        self.assertEqual(res.get("data").get("Keys")[0].get('Name'), key.get('data').get('ipnsName'))
+        self.assertEqual(res.get("data").get("Keys")[0].get('Id'), key.get('data').get('ipnsId'))