feat: implement GraphQL write interface with mutations (issue #10)
Some checks failed
Test Suite / performance-tests (push) Has been cancelled
Test Suite / unit-tests (3.11) (push) Has been cancelled
Test Suite / unit-tests (3.12) (push) Has been cancelled
Test Suite / integration-tests (push) Has been cancelled
Test Suite / e2e-tests (push) Has been cancelled
Test Suite / code-quality (push) Has been cancelled
Test Suite / security-scan (push) Has been cancelled
Test Suite / test-summary (push) Has been cancelled

Added comprehensive GraphQL mutations for CRUD operations on markdown files and schemas.

Key features:
- Complete mutation schema with structured payload types
- Markdown file mutations: add, update with front matter support
- Schema mutations: add, update, delete with JSON validation
- CLI integration with `graphql-mutate` command
- Comprehensive error handling and validation
- Full test coverage with 24 test cases
- Updated documentation with mutation examples and API usage

Resolves issue #10: Expose a GraphQL Write Interface

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-10-03 16:48:03 +02:00
parent d4e5992213
commit 2a15dde228
6 changed files with 2084 additions and 8 deletions

View File

@@ -435,6 +435,278 @@ class Query(QueryType):
return snippet
class Mutation:
"""GraphQL mutation resolver implementation."""
def __init__(self):
"""Initialize mutation resolver."""
self.resolver = MarkiTectResolver(get_default_database_path())
def resolve_add_markdown_file(self, info, filename, content):
"""Add a new markdown file to the database."""
try:
# Store the file using the database manager
file_id = self.resolver.db_manager.store_markdown_file(filename, content)
if file_id:
# Retrieve the created file
conn = self.resolver.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM markdown_files WHERE id = ?", (file_id,))
row = cursor.fetchone()
conn.close()
if row:
data = self.resolver.row_to_dict(cursor, row)
# Parse front matter JSON
if data['front_matter']:
try:
data['front_matter_raw'] = json.loads(data['front_matter'])
except json.JSONDecodeError:
data['front_matter_raw'] = {}
else:
data['front_matter_raw'] = {}
from .schema import AddMarkdownFilePayload, MarkdownFile
return AddMarkdownFilePayload(
markdown_file=MarkdownFile(**data),
success=True,
errors=[]
)
from .schema import AddMarkdownFilePayload
return AddMarkdownFilePayload(
markdown_file=None,
success=False,
errors=["Failed to store markdown file"]
)
except Exception as e:
from .schema import AddMarkdownFilePayload
return AddMarkdownFilePayload(
markdown_file=None,
success=False,
errors=[str(e)]
)
def resolve_update_markdown_file(self, info, id, content=None):
"""Update an existing markdown file."""
try:
if not content:
from .schema import UpdateMarkdownFilePayload
return UpdateMarkdownFilePayload(
markdown_file=None,
success=False,
errors=["Content is required for update"]
)
conn = self.resolver.get_connection()
cursor = conn.cursor()
# Check if file exists
cursor.execute("SELECT filename FROM markdown_files WHERE id = ?", (id,))
row = cursor.fetchone()
if not row:
conn.close()
from .schema import UpdateMarkdownFilePayload
return UpdateMarkdownFilePayload(
markdown_file=None,
success=False,
errors=[f"Markdown file with ID {id} not found"]
)
filename = row[0]
conn.close()
# Update using store_markdown_file (it handles front matter parsing)
file_id = self.resolver.db_manager.store_markdown_file(filename, content)
if file_id:
# Retrieve the updated file
conn = self.resolver.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM markdown_files WHERE filename = ?", (filename,))
row = cursor.fetchone()
conn.close()
if row:
data = self.resolver.row_to_dict(cursor, row)
# Parse front matter JSON
if data['front_matter']:
try:
data['front_matter_raw'] = json.loads(data['front_matter'])
except json.JSONDecodeError:
data['front_matter_raw'] = {}
else:
data['front_matter_raw'] = {}
from .schema import UpdateMarkdownFilePayload, MarkdownFile
return UpdateMarkdownFilePayload(
markdown_file=MarkdownFile(**data),
success=True,
errors=[]
)
from .schema import UpdateMarkdownFilePayload
return UpdateMarkdownFilePayload(
markdown_file=None,
success=False,
errors=["Failed to update markdown file"]
)
except Exception as e:
from .schema import UpdateMarkdownFilePayload
return UpdateMarkdownFilePayload(
markdown_file=None,
success=False,
errors=[str(e)]
)
def resolve_add_schema(self, info, filename, schema_content):
"""Add a new JSON schema to the database."""
try:
# Store the schema using the database manager
schema_id = self.resolver.db_manager.store_schema_file(filename, schema_content)
if schema_id:
# Retrieve the created schema
conn = self.resolver.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM schemas WHERE id = ?", (schema_id,))
row = cursor.fetchone()
conn.close()
if row:
data = self.resolver.row_to_dict(cursor, row)
# Parse schema content JSON
if data['schema_content']:
try:
data['schema_content'] = json.loads(data['schema_content'])
except json.JSONDecodeError:
data['schema_content'] = {}
from .schema import AddSchemaPayload, Schema
return AddSchemaPayload(
schema=Schema(**data),
success=True,
errors=[]
)
from .schema import AddSchemaPayload
return AddSchemaPayload(
schema=None,
success=False,
errors=["Failed to store schema"]
)
except Exception as e:
from .schema import AddSchemaPayload
return AddSchemaPayload(
schema=None,
success=False,
errors=[str(e)]
)
def resolve_update_schema(self, info, id, schema_content=None):
"""Update an existing JSON schema."""
try:
if not schema_content:
from .schema import UpdateSchemaPayload
return UpdateSchemaPayload(
schema=None,
success=False,
errors=["Schema content is required for update"]
)
conn = self.resolver.get_connection()
cursor = conn.cursor()
# Check if schema exists
cursor.execute("SELECT filename FROM schemas WHERE id = ?", (id,))
row = cursor.fetchone()
if not row:
conn.close()
from .schema import UpdateSchemaPayload
return UpdateSchemaPayload(
schema=None,
success=False,
errors=[f"Schema with ID {id} not found"]
)
filename = row[0]
conn.close()
# Update using store_schema_file
schema_id = self.resolver.db_manager.store_schema_file(filename, schema_content)
if schema_id:
# Retrieve the updated schema
conn = self.resolver.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM schemas WHERE filename = ?", (filename,))
row = cursor.fetchone()
conn.close()
if row:
data = self.resolver.row_to_dict(cursor, row)
# Parse schema content JSON
if data['schema_content']:
try:
data['schema_content'] = json.loads(data['schema_content'])
except json.JSONDecodeError:
data['schema_content'] = {}
from .schema import UpdateSchemaPayload, Schema
return UpdateSchemaPayload(
schema=Schema(**data),
success=True,
errors=[]
)
from .schema import UpdateSchemaPayload
return UpdateSchemaPayload(
schema=None,
success=False,
errors=["Failed to update schema"]
)
except Exception as e:
from .schema import UpdateSchemaPayload
return UpdateSchemaPayload(
schema=None,
success=False,
errors=[str(e)]
)
def resolve_delete_schema(self, info, filename):
"""Delete a JSON schema from the database."""
try:
# Delete using the database manager
success = self.resolver.db_manager.delete_schema_file(filename)
from .schema import DeleteSchemaPayload
if success:
return DeleteSchemaPayload(
success=True,
deleted_filename=filename,
errors=[]
)
else:
return DeleteSchemaPayload(
success=False,
deleted_filename=None,
errors=[f"Failed to delete schema: {filename}"]
)
except Exception as e:
from .schema import DeleteSchemaPayload
return DeleteSchemaPayload(
success=False,
deleted_filename=None,
errors=[str(e)]
)
def get_default_database_path():
"""Get default database path for GraphQL resolvers."""
import os