Merge pull request #2 from Kayrnt/README

Rename ReadMe.md to README.md & Lint/Format

Author: bazzazzadeh

ReadMe.md renamed to README.md

@@ -15,7 +15,6 @@
 from dotenv import find_dotenv, load_dotenv
 
 
-
 def get_env(name: str) -> str:
     value = os.getenv(name)
     if not value:
@@ -43,7 +42,9 @@ async def find_store(client: DeltaStreamClient, store_name: str):
         print("Available stores:")
         for store in stores:
             print(f"  - {store.name}")
-        print("\nPlease update STORE_NAME in .env to match one of the available stores.")
+        print(
+            "\nPlease update STORE_NAME in .env to match one of the available stores."
+        )
         sys.exit(1)
 
     print(f"\n✅ Found configured store: {target_store.name}")
@@ -84,18 +85,26 @@ async def ensure_entity(client: DeltaStreamClient, entity_name: str, target_stor
                     "kafka.topic.partitions": 1,
                     "kafka.topic.replicas": 1,
                     "kafka.topic.retention.ms": "604800000",  # 7 days
-                    "kafka.topic.segment.ms": "86400000"      # 1 day
-                }
+                    "kafka.topic.segment.ms": "86400000",  # 1 day
+                },
+            )
+            print(
+                f"Creating entity with params: name='{entity_name}', store='{target_store.name}', params={create_params.params}"
             )
-            print(f"Creating entity with params: name='{entity_name}', store='{target_store.name}', params={create_params.params}")
             await client.entities.create(params=create_params)
-            print(f"Entity '{entity_name}' created successfully in store '{target_store.name}'")
+            print(
+                f"Entity '{entity_name}' created successfully in store '{target_store.name}'"
+            )
         except Exception as e:
             error_msg = str(e)
             if "already exists" in error_msg:
-                print(f"Entity '{entity_name}' already exists (topic level), continuing...")
+                print(
+                    f"Entity '{entity_name}' already exists (topic level), continuing..."
+                )
             elif "schema not found" in error_msg:
-                print(f"Entity '{entity_name}' created successfully but retrieval failed, continuing...")
+                print(
+                    f"Entity '{entity_name}' created successfully but retrieval failed, continuing..."
+                )
             else:
                 print(f"Could not create entity using SDK: {e}")
                 sys.exit(1)
@@ -112,13 +121,15 @@ async def insert_data(client: DeltaStreamClient, entity_name: str, store_name: s
         {"viewtime": 1753311018650, "userid": "User_4", "pageid": "Page_2"},
     ]
 
-    print(f"Attempting to insert {len(sample_data)} records into entity '{entity_name}' in store '{store_name}'...")
-    print(f"Generated SQL will be: INSERT INTO ENTITY \"{entity_name}\" IN STORE \"{store_name}\" VALUE ('{{\"viewtime\": 1753311018649,\"userid\": \"User_3\",\"pageid\": \"Page_1\"}}')")
+    print(
+        f"Attempting to insert {len(sample_data)} records into entity '{entity_name}' in store '{store_name}'..."
+    )
+    print(
+        f'Generated SQL will be: INSERT INTO ENTITY "{entity_name}" IN STORE "{store_name}" VALUE (\'{{"viewtime": 1753311018649,"userid": "User_3","pageid": "Page_1"}}\')'
+    )
 
     await client.entities.insert_values(
-        name=entity_name,
-        values=sample_data,
-        store=store_name
+        name=entity_name, values=sample_data, store=store_name
     )
     print("Insert completed successfully.")
 

examples/insert_entity/src/insert_entity_example.py

@@ -61,13 +61,9 @@ def __init__(self) -> None:
     def validate(self) -> None:
         """Validate that required configuration is present."""
         if not self.auth_token:
-            raise ValueError(
-                "DELTASTREAM_TOKEN environment variable is required"
-            )
+            raise ValueError("DELTASTREAM_TOKEN environment variable is required")
         if not self.organization_id:
-            raise ValueError(
-                "DELTASTREAM_ORG_ID environment variable is required"
-            )
+            raise ValueError("DELTASTREAM_ORG_ID environment variable is required")
 
         # Check token validity
         self._validate_token()

examples/list_entity/src/utils/config.py

@@ -26,6 +26,6 @@
     "InvalidConfiguration",
     "ConnectionError",
     "models",
-    "resources", 
+    "resources",
     "exceptions",
 ]

src/deltastream_sdk/__init__.py

@@ -251,30 +251,38 @@ async def query_sql(self, sql: str) -> List[Dict[str, Any]]:
     async def use_database(self, database: str) -> None:
         """Switch to a different database."""
         # Use DATABASE keyword for DeltaStream syntax
-        escaped_db = database if database.startswith('"') and database.endswith('"') else f'"{database}"'
-        sql = f'USE DATABASE {escaped_db}'
+        escaped_db = (
+            database
+            if database.startswith('"') and database.endswith('"')
+            else f'"{database}"'
+        )
+        sql = f"USE DATABASE {escaped_db}"
         await self.execute_sql(sql)
-        
+
         # Update the current database in memory
         self._current_database = database
 
     async def use_store(self, store: str) -> None:
         """Switch to a different store."""
         # Use STORE keyword for DeltaStream syntax
-        escaped_store = store if store.startswith('"') and store.endswith('"') else f'"{store}"'
-        sql = f'USE STORE {escaped_store}'
+        escaped_store = (
+            store if store.startswith('"') and store.endswith('"') else f'"{store}"'
+        )
+        sql = f"USE STORE {escaped_store}"
         await self.execute_sql(sql)
-        
+
         # Update the current store in memory
         self._current_store = store
 
     async def use_schema(self, schema: str) -> None:
         """Switch to a different schema."""
         # Use SCHEMA keyword for DeltaStream syntax
-        escaped_schema = schema if schema.startswith('"') and schema.endswith('"') else f'"{schema}"'
-        sql = f'USE SCHEMA {escaped_schema}'
+        escaped_schema = (
+            schema if schema.startswith('"') and schema.endswith('"') else f'"{schema}"'
+        )
+        sql = f"USE SCHEMA {escaped_schema}"
         await self.execute_sql(sql)
-        
+
         # Update the current schema in memory
         self._current_schema = schema
 
@@ -283,74 +291,74 @@ async def get_current_database(self) -> str:
         # If we have a cached current database, return it
         if self._current_database:
             return self._current_database
-        
+
         # Otherwise, query LIST DATABASES to find the default database
         try:
             databases = await self.databases.list()
             for db in databases:
                 # Look for the default database
-                if hasattr(db, 'is_default') and db.is_default:
+                if hasattr(db, "is_default") and db.is_default:
                     self._current_database = db.name
                     return db.name
-            
+
             # If no default found, return the first database if any exist
             if databases:
                 self._current_database = databases[0].name
                 return databases[0].name
         except Exception:
             # If list databases fails, return empty string
             pass
-        
+
         return ""
 
     async def get_current_store(self) -> str:
         """Get the current store."""
         # If we have a cached current store, return it
         if self._current_store:
             return self._current_store
-        
+
         # Otherwise, query LIST STORES to find the default store
         try:
             stores = await self.stores.list()
             for store in stores:
                 # Look for the default store
-                if hasattr(store, 'is_default') and store.is_default:
+                if hasattr(store, "is_default") and store.is_default:
                     self._current_store = store.name
                     return store.name
-            
+
             # If no default found, return the first store if any exist
             if stores:
                 self._current_store = stores[0].name
                 return stores[0].name
         except Exception:
             # If list stores fails, return empty string
             pass
-        
+
         return ""
 
     async def get_current_schema(self) -> str:
         """Get the current schema."""
         # If we have a cached current schema, return it
         if self._current_schema:
             return self._current_schema
-        
+
         # Otherwise, query LIST SCHEMAS to find the default schema
         try:
             schemas = await self.schemas.list()
             for schema in schemas:
                 # Look for the default schema
-                if hasattr(schema, 'is_default') and schema.is_default:
+                if hasattr(schema, "is_default") and schema.is_default:
                     self._current_schema = schema.name
                     return schema.name
-            
+
             # If no default found, return the first schema if any exist
             if schemas:
                 self._current_schema = schemas[0].name
                 return schemas[0].name
         except Exception:
             # If list schemas fails, return empty string
             pass
-        
+
         return ""
 
     # Context manager support

src/deltastream_sdk/client.py

@@ -11,7 +11,9 @@ class Entity(BaseModel):
 
     entity_type: Optional[str] = None
     schema_definition: Optional[str] = None
-    is_leaf: Optional[bool] = None  # Whether this entity is a leaf (can't contain other entities)
+    is_leaf: Optional[bool] = (
+        None  # Whether this entity is a leaf (can't contain other entities)
+    )
 
 
 @dataclass

src/deltastream_sdk/models/entities.py

@@ -40,4 +40,4 @@ class SchemaCreateParams:
     """Parameters for creating a schema."""
 
     name: str
-    comment: Optional[str] = None
+    comment: Optional[str] = None

src/deltastream_sdk/models/schemas.py

@@ -2,7 +2,6 @@
 
 from .base import BaseResourceManager
 from ..models.databases import Database, DatabaseCreateParams
-from ..exceptions import ResourceNotFound, SQLError
 
 
 class DatabaseManager(BaseResourceManager[Database]):

src/deltastream_sdk/resources/databases.py

@@ -9,7 +9,6 @@
     EntityCreateParams,
     EntityUpdateParams,
 )
-from ..models.base import WithClause
 
 
 class EntityManager(BaseResourceManager[Entity]):
@@ -33,20 +32,20 @@ def _get_create_sql(self, **params) -> str:
 
         name = self._escape_identifier(create_params.name)
         sql = f"CREATE ENTITY {name}"
-        
+
         if create_params.store:
             escaped_store = self._escape_identifier(create_params.store)
             sql += f" IN STORE {escaped_store}"
-        
+
         with_parts = []
-        
+
         if create_params.params:
             for key, value in create_params.params.items():
                 with_parts.append(f"'{key}' = {self._escape_string(str(value))}")
-        
+
         if with_parts:
             sql += f" WITH ({', '.join(with_parts)})"
-        
+
         return sql
 
     def _get_update_sql(self, name: str, **params) -> str:
@@ -92,14 +91,14 @@ def to_json_string(record: Union[str, Dict[str, Any]]) -> str:
             single = values[0]
             json_str = self._escape_string(to_json_string(single))
             sql = f"INSERT INTO ENTITY {escaped_name}"
-            
+
             # Add IN STORE clause if store is provided
             if store:
                 escaped_store = self._escape_identifier(store)
                 sql += f" IN STORE {escaped_store}"
-            
+
             sql += f" VALUE({json_str})"
-            
+
             # Add WITH clause if we have with_params
             if with_params:
                 with_parts = []
@@ -110,21 +109,23 @@ def to_json_string(record: Union[str, Dict[str, Any]]) -> str:
             for record in values:
                 json_str = self._escape_string(to_json_string(record))
                 single_sql = f"INSERT INTO ENTITY {escaped_name}"
-                
+
                 # Add IN STORE clause if store is provided
                 if store:
                     escaped_store = self._escape_identifier(store)
                     single_sql += f" IN STORE {escaped_store}"
-                
+
                 single_sql += f" VALUE({json_str})"
-                
+
                 # Add WITH clause if we have with_params
                 if with_params:
                     with_parts = []
                     for key, value in with_params.items():
-                        with_parts.append(f"'{key}' = {self._escape_string(str(value))}")
+                        with_parts.append(
+                            f"'{key}' = {self._escape_string(str(value))}"
+                        )
                     single_sql += f" WITH ({', '.join(with_parts)})"
-                    
+
                 await self._execute_sql(single_sql)
             return
 
@@ -146,17 +147,17 @@ async def list_entities(
             List of Entity objects representing the entities
         """
         sql = "LIST ENTITIES"
-        
+
         # Add IN clause for entity path if provided
         if entity_path:
             escaped_path = self._escape_identifier(entity_path)
             sql += f" IN {escaped_path}"
-        
+
         # Add IN STORE clause if provided
         if store:
             escaped_store = self._escape_identifier(store)
             sql += f" IN STORE {escaped_store}"
-        
+
         try:
             results = await self._query_sql(sql)
             entities = []
@@ -171,9 +172,10 @@ async def list_entities(
                     entity.is_leaf = result["Is Leaf"]
                 elif "is_leaf" in result:
                     entity.is_leaf = result["is_leaf"]
-                    
+
                 entities.append(entity)
             return entities
         except Exception as e:
             from ..exceptions import SQLError
+
             raise SQLError(f"Failed to list entities: {e}") from e

src/deltastream_sdk/resources/entities.py

@@ -47,4 +47,4 @@ def _get_update_sql(self, name: str, **params) -> str:
     def _get_delete_sql(self, name: str, **params) -> str:
         """Generate SQL for deleting a schema."""
         escaped_name = self._escape_identifier(name)
-        return f"DROP SCHEMA {escaped_name}"
+        return f"DROP SCHEMA {escaped_name}"