Add column types system, refs #2664 · #2667

+class ColumnType:+    """+    Base class for column types.+ +    Subclasses must define ``name`` and ``description`` as class attributes:+ +    - ``name``: Unique identifier string. Lowercase, no spaces.+      Examples: "markdown", "file", "email", "url", "point", "image".+    - ``description``: Human-readable label for admin UI dropdowns.+      Examples: "Markdown text", "File reference", "Email address".+ +    Instantiate with an optional ``config`` dict to bind per-column+    configuration::+ +        ct = MyColumnType(config={"key": "value"})+        ct.config  # {"key": "value"}+    """+ +    name: str+    description: str+ +    def __init__(self, config=None):+        self.config = config+ +    async def render_cell(self, value, column, table, database, datasette, request):+        """+        Return an HTML string to render this cell value, or None to+        fall through to the default render_cell plugin hook chain.+        """+        return None+ +    async def validate(self, value, datasette):+        """+        Validate a value before it is written. Return None if valid,+        or a string error message if invalid.+        """+        return None+ +    async def transform_value(self, value, datasette):+        """+        Transform a value before it appears in JSON API output.+        Return the transformed value. Default: return unchanged.+        """+        return value

     @hookspec-def render_cell(row, value, column, table, pks, database, datasette, request):+def render_cell(+    row,+    value,+    column,+    table,+    pks,+    database,+    datasette,+    request,+    column_type,+):     """Customize rendering of HTML table cell values"""    ⋯     """Register actions: returns a list of datasette.permission.Action objects"""    +@hookspec+def register_column_types(datasette):+    """Return a list of ColumnType instances"""+ +  @hookspec def register_routes(datasette):     """Register URL routes: return a list of (regex, view_function) pairs"""

+import json+import re+ ⋯+import markupsafe+ ⋯+from datasette import hookimpl+from datasette.column_types import ColumnType+ + ⋯+class UrlColumnType(ColumnType):+    name = "url"+    description = "URL"+ +    async def render_cell(self, value, column, table, database, datasette, request):+        if not value or not isinstance(value, str):+            return None+        escaped = markupsafe.escape(value.strip())+        return markupsafe.Markup(f'<a href="{escaped}">{escaped}</a>')+ +    async def validate(self, value, datasette):+        if value is None or value == "":+            return None+        if not isinstance(value, str):+            return "URL must be a string"+        if not re.match(r"^https?://\S+$", value.strip()):+            return "Invalid URL"+        return None+ + ⋯+class EmailColumnType(ColumnType):+    name = "email"+    description = "Email address"+ +    async def render_cell(self, value, column, table, database, datasette, request):+        if not value or not isinstance(value, str):+            return None+        escaped = markupsafe.escape(value.strip())+        return markupsafe.Markup(f'<a href="mailto:{escaped}">{escaped}</a>')+ +    async def validate(self, value, datasette):+        if value is None or value == "":+            return None+        if not isinstance(value, str):+            return "Email must be a string"+        if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", value.strip()):+            return "Invalid email address"+        return None+ + ⋯+class JsonColumnType(ColumnType):+    name = "json"+    description = "JSON data"+ +    async def render_cell(self, value, column, table, database, datasette, request):+        if value is None:+            return None+        try:+            parsed = json.loads(value) if isinstance(value, str) else value+            formatted = json.dumps(parsed, indent=2)+            escaped = markupsafe.escape(formatted)+            return markupsafe.Markup(f"<pre>{escaped}</pre>")+        except (json.JSONDecodeError, TypeError):+            return None+ +    async def validate(self, value, datasette):+        if value is None or value == "":+            return None+        if isinstance(value, str):+            try:+                json.loads(value)+            except json.JSONDecodeError:+                return "Invalid JSON"+        return None+ + ⋯+@hookimpl+def register_column_types(datasette):+    return [UrlColumnType, EmailColumnType, JsonColumnType]

     "datasette.default_permissions",     "datasette.default_permissions.tokens",     "datasette.default_actions",+    "datasette.default_column_types",     "datasette.default_magic_parameters",     "datasette.blob_renderer",     "datasette.default_menu_links",

             value text,             unique(database_name, resource_name, column_name, key)         );+ +        CREATE TABLE IF NOT EXISTS column_types (+            database_name TEXT NOT NULL,+            resource_name TEXT NOT NULL,+            column_name TEXT NOT NULL,+            column_type TEXT NOT NULL,+            config TEXT,+            PRIMARY KEY (database_name, resource_name, column_name)+        );             """))    

         self.immutables = set(immutables or [])         self.databases = collections.OrderedDict()         self.actions = {}  # .invoke_startup() will populate this+        self._column_types = {}  # .invoke_startup() will populate this         try:             self._refresh_schemas_lock = asyncio.Lock()         except RuntimeError as rex:⋯                         action_abbrs[action.abbr] = action                     self.actions[action.name] = action  +        # Register column types (classes, not instances)+        self._column_types = {}+        for hook in pm.hook.register_column_types(datasette=self):+            if hook:+                for ct_cls in hook:+                    if ct_cls.name in self._column_types:+                        raise StartupError(f"Duplicate column type name: {ct_cls.name}")+                    self._column_types[ct_cls.name] = ct_cls+          for hook in pm.hook.prepare_jinja2_environment(             env=self._jinja_env, datasette=self         ):⋯             await await_me_maybe(hook)         # Ensure internal tables and metadata are populated before startup hooks         await self._refresh_schemas()+        # Load column_types from config into internal DB+        await self._apply_column_types_config()         for hook in pm.hook.startup(datasette=self):             await await_me_maybe(hook)         self._startup_invoked = True

             [database_name, resource_name, column_name, key, value],         )  +    # Column types API+ +    async def _apply_column_types_config(self):+        """Load column_types from datasette.json config into the internal DB."""+        import logging+ +        for db_name, db_conf in (self.config or {}).get("databases", {}).items():+            for table_name, table_conf in db_conf.get("tables", {}).items():+                for col_name, ct in table_conf.get("column_types", {}).items():+                    if isinstance(ct, str):+                        col_type, config = ct, None+                    else:+                        col_type = ct["type"]+                        config = ct.get("config")+                    if col_type not in self._column_types:+                        logging.warning(+                            "column_types config references unknown type %r "+                            "for %s.%s.%s",+                            col_type,+                            db_name,+                            table_name,+                            col_name,+                        )+                    await self.set_column_type(+                        db_name, table_name, col_name, col_type, config+                    )+ +    async def get_column_type(self, database: str, resource: str, column: str):+        """+        Return a ColumnType instance (with config baked in) for a specific+        column, or None if no column type is assigned.+        """+        row = await self.get_internal_database().execute(+            "SELECT column_type, config FROM column_types "+            "WHERE database_name = ? AND resource_name = ? AND column_name = ?",+            [database, resource, column],+        )+        rows = row.rows+        if not rows:+            return None+        ct_name, config = rows[0]+        ct_cls = self._column_types.get(ct_name)+        if ct_cls is None:+            return None+        return ct_cls(config=json.loads(config) if config else None)+ +    async def get_column_types(self, database: str, resource: str) -> dict:+        """+        Return {column_name: ColumnType instance (with config)}+        for all columns with assigned types on the given resource.+        """+        rows = await self.get_internal_database().execute(+            "SELECT column_name, column_type, config FROM column_types "+            "WHERE database_name = ? AND resource_name = ?",+            [database, resource],+        )+        result = {}+        for row in rows.rows:+            col_name, ct_name, config = row+            ct_cls = self._column_types.get(ct_name)+            if ct_cls is not None:+                result[col_name] = ct_cls(config=json.loads(config) if config else None)+        return result+ +    async def set_column_type(+        self,+        database: str,+        resource: str,+        column: str,+        column_type: str,+        config: dict = None,+    ) -> None:+        """Assign a column type. Overwrites any existing assignment."""+        await self.get_internal_database().execute_write(+            """INSERT OR REPLACE INTO column_types+               (database_name, resource_name, column_name, column_type, config)+               VALUES (?, ?, ?, ?, ?)""",+            [+                database,+                resource,+                column,+                column_type,+                json.dumps(config) if config else None,+            ],+        )+ +    async def remove_column_type(+        self, database: str, resource: str, column: str+    ) -> None:+        """Remove a column type assignment."""+        await self.get_internal_database().execute_write(+            "DELETE FROM column_types "+            "WHERE database_name = ? AND resource_name = ? AND column_name = ?",+            [database, resource, column],+        )+      def get_internal_database(self):         return self._internal_database  

         )     )  +    # Look up column types for this table+    column_types_map = await datasette.get_column_types(database_name, table_name)+      column_details = {         col.name: col for col in await db.table_column_details(table_name)     }⋯         else:             type_ = column_details[r[0]].type             notnull = column_details[r[0]].notnull-        columns.append(-            {-                "name": r[0],-                "sortable": r[0] in sortable_columns,-                "is_pk": r[0] in pks_for_display,-                "type": type_,-                "notnull": notnull,-                "description": column_descriptions.get(r[0]),-            }-        )+        col_dict = {+            "name": r[0],+            "sortable": r[0] in sortable_columns,+            "is_pk": r[0] in pks_for_display,+            "type": type_,+            "notnull": notnull,+            "description": column_descriptions.get(r[0]),+            "column_type": None,+            "column_type_config": None,+        }+        ct = column_types_map.get(r[0])+        if ct:+            col_dict["column_type"] = ct.name+            col_dict["column_type_config"] = ct.config+        columns.append(col_dict)       column_to_foreign_key_table = {         fk["column"]: fk["other_table"]⋯                 # already shown in the link column.                 continue  -            # First let the plugins have a go+            # First try column type render_cell, then plugins             # pylint: disable=no-member             plugin_display_value = None⋯             # pylint: disable=no-member             plugin_display_value = None-            for candidate in pm.hook.render_cell(-                row=row,-                value=value,-                column=column,-                table=table_name,-                pks=pks_for_display,-                database=database_name,-                datasette=datasette,-                request=request,-            ):-                candidate = await await_me_maybe(candidate)+            ct = column_types_map.get(column)+            if ct:+                candidate = await ct.render_cell(+                    value=value,+                    column=column,+                    table=table_name,+                    database=database_name,+                    datasette=datasette,+                    request=request,+                )                 if candidate is not None:                     plugin_display_value = candidate⋯                 if candidate is not None:                     plugin_display_value = candidate-                    break+            if plugin_display_value is None:+                for candidate in pm.hook.render_cell(+                    row=row,+                    value=value,+                    column=column,+                    table=table_name,+                    pks=pks_for_display,+                    database=database_name,+                    datasette=datasette,+                    request=request,+                    column_type=ct,+                ):+                    candidate = await await_me_maybe(candidate)+                    if candidate is not None:+                        plugin_display_value = candidate+                        break             if plugin_display_value:                 display_value = plugin_display_value             elif isinstance(value, bytes):

     async def extra_render_cell():         "Rendered HTML for each cell using the render_cell plugin hook"         pks_for_display = pks if pks else (["rowid"] if not is_view else [])-        columns = [col[0] for col in results.description]+        col_names = [col[0] for col in results.description]+        ct_map = await datasette.get_column_types(database_name, table_name)         rendered_rows = []         for row in rows:             rendered_row = {}⋯         rendered_rows = []         for row in rows:             rendered_row = {}-            for value, column in zip(row, columns):-                # Call render_cell plugin hook+            for value, column in zip(row, col_names):+                ct = ct_map.get(column)                 plugin_display_value = None⋯                 plugin_display_value = None-                for candidate in pm.hook.render_cell(-                    row=row,-                    value=value,-                    column=column,-                    table=table_name,-                    pks=pks_for_display,-                    database=database_name,-                    datasette=datasette,-                    request=request,-                ):-                    candidate = await await_me_maybe(candidate)+                # Try column type render_cell first+                if ct:+                    candidate = await ct.render_cell(+                        value=value,+                        column=column,+                        table=table_name,+                        database=database_name,+                        datasette=datasette,+                        request=request,+                    )                     if candidate is not None:                         plugin_display_value = candidate⋯                     if candidate is not None:                         plugin_display_value = candidate-                        break+                if plugin_display_value is None:+                    for candidate in pm.hook.render_cell(+                        row=row,+                        value=value,+                        column=column,+                        table=table_name,+                        pks=pks_for_display,+                        database=database_name,+                        datasette=datasette,+                        request=request,+                        column_type=ct,+                    ):+                        candidate = await await_me_maybe(candidate)+                        if candidate is not None:+                            plugin_display_value = candidate+                            break                 if plugin_display_value:                     rendered_row[column] = str(plugin_display_value)             rendered_rows.append(rendered_row)

           if "render_cell" in extras:             # Call render_cell plugin hook for each cell+            ct_map = await self.ds.get_column_types(database, table)             rendered_rows = []             for row in rows:                 rendered_row = {}⋯             for row in rows:                 rendered_row = {}                 for value, column in zip(row, columns):-                    # Call render_cell plugin hook+                    ct = ct_map.get(column)                     plugin_display_value = None⋯                     plugin_display_value = None-                    for candidate in pm.hook.render_cell(-                        row=row,-                        value=value,-                        column=column,-                        table=table,-                        pks=resolved.pks,-                        database=database,-                        datasette=self.ds,-                        request=request,-                    ):-                        candidate = await await_me_maybe(candidate)+                    # Try column type render_cell first+                    if ct:+                        candidate = await ct.render_cell(+                            value=value,+                            column=column,+                            table=table,+                            database=database,+                            datasette=self.ds,+                            request=request,+                        )                         if candidate is not None:                             plugin_display_value = candidate⋯                         if candidate is not None:                             plugin_display_value = candidate-                            break+                    if plugin_display_value is None:+                        for candidate in pm.hook.render_cell(+                            row=row,+                            value=value,+                            column=column,+                            table=table,+                            pks=resolved.pks,+                            database=database,+                            datasette=self.ds,+                            request=request,+                            column_type=ct,+                        ):+                            candidate = await await_me_maybe(candidate)+                            if candidate is not None:+                                plugin_display_value = candidate+                                break                     if plugin_display_value:                         rendered_row[column] = str(plugin_display_value)                 rendered_rows.append(rendered_row)

                 database=database,                 datasette=datasette,                 request=request,+                column_type=None,             ):                 candidate = await await_me_maybe(candidate)                 if candidate is not None:

             "params": params,         }  +    async def extra_column_types():+        "Column type assignments for this table"+        ct_map = await datasette.get_column_types(database_name, table_name)+        return {+            col_name: {+                "type": ct.name,+                "config": ct.config,+            }+            for col_name, ct in ct_map.items()+        }+      async def extra_metadata():         "Metadata about the table and database"         tablemetadata = await datasette.get_resource_metadata(database_name, table_name)⋯         extra_debug,         extra_request,         extra_query,+        extra_column_types,         extra_metadata,         extra_extras,         extra_database,

         }     )     raw_sqlite_rows = rows[:page_size]-    data["rows"] = [dict(r) for r in raw_sqlite_rows]+    # Apply transform_value for columns with assigned types+    ct_map = await datasette.get_column_types(database_name, table_name)+    transformed_rows = []+    for r in raw_sqlite_rows:+        row_dict = dict(r)+        for col_name, ct in ct_map.items():+            if col_name in row_dict:+                row_dict[col_name] = await ct.transform_value(+                    row_dict[col_name], datasette+                )+        transformed_rows.append(row_dict)+    data["rows"] = transformed_rows       if context_for_html_hack:         data.update(extra_context_from_filters)

         )    +async def _validate_column_types(datasette, database_name, table_name, rows):+    """Validate row values against assigned column types. Returns list of error strings."""+    ct_map = await datasette.get_column_types(database_name, table_name)+    if not ct_map:+        return []+    errors = []+    for row in rows:+        for col_name, ct in ct_map.items():+            if col_name not in row:+                continue+            error = await ct.validate(row[col_name], datasette)+            if error:+                errors.append(f"{col_name}: {error}")+    return errors+ +  async def display_columns_and_rows(     datasette,     database_name,

         if errors:             return _error(errors, 400)  +        # Validate column types+        ct_errors = await _validate_column_types(+            self.ds, database_name, table_name, rows+        )+        if ct_errors:+            return _error(ct_errors, 400)+          num_rows = len(rows)           # No that we've passed pks to _validate_data it's safe to

           update = data["update"]  +        # Validate column types+        from datasette.views.table import _validate_column_types+ +        ct_errors = await _validate_column_types(+            self.ds, resolved.db.name, resolved.table, [update]+        )+        if ct_errors:+            return _error(ct_errors, 400)+          alter = data.get("alter")         if alter and not await self.ds.allowed(             action="alter-table",

 Any previous column-level metadata entry with the same ``key`` will be overwritten. Internally upserts the value into the  the ``metadata_columns`` table inside the :ref:`internal database <internals_internal>`.  +.. _datasette_column_types:+ ⋯+Column types+------------+ ⋯+Column types are stored in the ``column_types`` table in the :ref:`internal database <internals_internal>`. The following methods provide the API for reading and modifying column type assignments.+ ⋯+.. _datasette_get_column_type:+ ⋯+await .get_column_type(database, resource, column)+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~+ ⋯+``database`` - string+    The name of the database.+``resource`` - string+    The name of the table or view.+``column`` - string+    The name of the column.+ ⋯+Returns a :ref:`ColumnType <column_types>` subclass instance with ``.config`` populated for the specified column, or ``None`` if no column type is assigned.+ ⋯+.. code-block:: python+ +    ct = await datasette.get_column_type(+        "mydb", "mytable", "email_col"+    )+    if ct:+        print(ct.name)  # "email"+        print(ct.config)  # None or {...}+ ⋯+.. _datasette_get_column_types:+ ⋯+await .get_column_types(database, resource)+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~+ ⋯+``database`` - string+    The name of the database.+``resource`` - string+    The name of the table or view.+ ⋯+Returns a dictionary mapping column names to :ref:`ColumnType <column_types>` subclass instances (with ``.config`` populated) for all columns that have assigned types on the given resource.+ ⋯+.. code-block:: python+ +    ct_map = await datasette.get_column_types("mydb", "mytable")+    for col_name, ct in ct_map.items():+        print(col_name, ct.name, ct.config)+ ⋯+.. _datasette_set_column_type:+ ⋯+await .set_column_type(database, resource, column, column_type, config=None)+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~+ ⋯+``database`` - string+    The name of the database.+``resource`` - string+    The name of the table or view.+``column`` - string+    The name of the column.+``column_type`` - string+    The column type name to assign, e.g. ``"email"``.+``config`` - dict, optional+    Optional configuration dict for the column type.+ ⋯+Assigns a column type to a column. Overwrites any existing assignment for that column.+ ⋯+.. code-block:: python+ +    await datasette.set_column_type(+        "mydb",+        "mytable",+        "location",+        "point",+        config={"srid": 4326},+    )+ ⋯+.. _datasette_remove_column_type:+ ⋯+await .remove_column_type(database, resource, column)+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~+ ⋯+``database`` - string+    The name of the database.+``resource`` - string+    The name of the table or view.+``column`` - string+    The name of the column.+ ⋯+Removes the column type assignment for the specified column.+ ⋯+.. code-block:: python+ +    await datasette.remove_column_type(+        "mydb", "mytable", "location"+    )+  .. _datasette_add_database:   .add_database(db, name=None, route=None)

         value text,         unique(database_name, resource_name, column_name, key)     );+    CREATE TABLE column_types (+        database_name TEXT NOT NULL,+        resource_name TEXT NOT NULL,+        column_name TEXT NOT NULL,+        column_type TEXT NOT NULL,+        config TEXT,+        PRIMARY KEY (database_name, resource_name, column_name)+    );   .. [[[end]]]  

   .. _plugin_hook_render_cell:  -render_cell(row, value, column, table, pks, database, datasette, request)--------------------------------------------------------------------------+render_cell(row, value, column, table, pks, database, datasette, request, column_type)+--------------------------------------------------------------------------------------   Lets you customize the display of values within table cells in the HTML table view.  ⋯ ``request`` - :ref:`internals_request`     The current request object  +``column_type`` - :ref:`ColumnType <column_types>` subclass instance or None+    The :ref:`ColumnType <column_types>` subclass instance assigned to this column (with ``.config`` populated), or ``None`` if no column type is assigned. You can access ``column_type.name``, ``column_type.config``, etc.+ ⋯+If a column has a :ref:`column type <column_types>` assigned and that column type's ``render_cell`` method returns a non-``None`` value, it will take priority over this plugin hook.+  If your hook returns ``None``, it will be ignored. Use this to indicate that your hook is not able to custom render this particular value.   If the hook returns a string, that string will be rendered in the table cell.

   The permission system then uses this query along with rules from plugins to determine which documents each user can access, all efficiently in SQL rather than loading everything into Python.  +.. _plugin_register_column_types:+ ⋯+register_column_types(datasette)+--------------------------------+ ⋯+Return a list of :ref:`ColumnType <column_types>` **subclasses** (not instances) to register custom column types. Column types define how values in specific columns are rendered, validated, and transformed.+ ⋯+.. code-block:: python+ +    from datasette import hookimpl+    from datasette.column_types import ColumnType+    import markupsafe+ + +    class ColorColumnType(ColumnType):+        name = "color"+        description = "CSS color value"+ +        async def render_cell(+            self,+            value,+            column,+            table,+            database,+            datasette,+            request,+        ):+            if value:+                return markupsafe.Markup(+                    '<span style="background-color: {color}">'+                    "{color}</span>"+                ).format(color=markupsafe.escape(value))+            return None+ +        async def validate(self, value, datasette):+            if value and not value.startswith("#"):+                return "Color must start with #"+            return None+ +        async def transform_value(self, value, datasette):+            # Normalize to uppercase+            if isinstance(value, str):+                return value.upper()+            return value+ + +    @hookimpl+    def register_column_types(datasette):+        return [ColorColumnType]+ ⋯+Each ``ColumnType`` subclass must define the following class attributes:+ ⋯+``name`` - string+    Unique identifier for the column type, e.g. ``"color"``. Must be unique across all plugins.+ ⋯+``description`` - string+    Human-readable label, e.g. ``"CSS color value"``.+ ⋯+And the following methods, all optional:+ ⋯+``render_cell(self, value, column, table, database, datasette, request)``+    Return an HTML string to render this cell value, or ``None`` to fall through to the default ``render_cell`` plugin hook chain. When a column type provides rendering, it takes priority over the ``render_cell`` plugin hook.+ ⋯+``validate(self, value, datasette)``+    Validate a value before it is written via the insert, update, or upsert API endpoints. Return ``None`` if valid, or a string error message if invalid. Null values and empty strings skip validation.+ ⋯+``transform_value(self, value, datasette)``+    Transform a value before it appears in JSON API output. Return the transformed value. The default implementation returns the value unchanged.+ ⋯+Per-column configuration is available via ``self.config`` in all methods. When a column type is looked up for a specific column (via :ref:`get_column_type <datasette_get_column_type>` or :ref:`get_column_types <datasette_get_column_types>`), the returned instance has ``config`` set to the parsed JSON config dict for that column assignment, or ``None`` if no config was provided.+ ⋯+Column types are assigned to columns via the ``column_types`` key in :ref:`table configuration <metadata_tables>`:+ ⋯+.. code-block:: yaml+ +    databases:+      mydb:+        tables:+          mytable:+            column_types:+              bg_color: color+              highlight:+                type: color+                config:+                  format: rgb+ ⋯+Datasette includes three built-in column types: ``url``, ``email``, and ``json``.+  .. _plugin_asgi_wrapper:   asgi_wrapper(datasette)

                 "register_actions"             ]         },+        {+            "name": "datasette.default_column_types",+            "static": false,+            "templates": false,+            "version": null,+            "hooks": [+                "register_column_types"+            ]+        },         {             "name": "datasette.default_magic_parameters",             "static": false,

+import logging+ ⋯+from datasette.app import Datasette+from datasette.column_types import ColumnType+from datasette.hookspecs import hookimpl+from datasette.plugins import pm+from datasette.utils import sqlite3+from datasette.utils import StartupError+import markupsafe+import pytest+import time+ + ⋯+@pytest.fixture+def ds_ct(tmp_path_factory):+    db_directory = tmp_path_factory.mktemp("dbs")+    db_path = str(db_directory / "data.db")+    db = sqlite3.connect(str(db_path))+    db.execute("vacuum")+    db.execute(+        "create table posts (id integer primary key, title text, body text, "+        "author_email text, website text, metadata text)"+    )+    db.execute(+        "insert into posts values (1, 'Hello', '# World', 'test@example.com', "+        "'https://example.com', '{\"key\": \"value\"}')"+    )+    db.commit()+    ds = Datasette(+        [db_path],+        config={+            "databases": {+                "data": {+                    "tables": {+                        "posts": {+                            "column_types": {+                                "body": "markdown",+                                "author_email": "email",+                                "website": "url",+                                "metadata": "json",+                            }+                        }+                    }+                }+            }+        },+    )+    ds.root_enabled = True+    yield ds+    db.close()+    for database in ds.databases.values():+        if not database.is_memory:+            database.close()+ + ⋯+def write_token(ds, actor_id="root", permissions=None):+    to_sign = {"a": actor_id, "token": "dstok", "t": int(time.time())}+    if permissions:+        to_sign["_r"] = {"a": permissions}+    return "dstok_{}".format(ds.sign(to_sign, namespace="token"))+ + ⋯+def _headers(token):+    return {+        "Authorization": "Bearer {}".format(token),+        "Content-Type": "application/json",+    }+ + ⋯+# --- Internal DB and config loading ---+ + ⋯+@pytest.mark.asyncio+async def test_column_types_table_created(ds_ct):+    await ds_ct.invoke_startup()+    internal = ds_ct.get_internal_database()+    result = await internal.execute(+        "SELECT name FROM sqlite_master WHERE type='table' AND name='column_types'"+    )+    assert len(result.rows) == 1+ + ⋯+@pytest.mark.asyncio+async def test_config_loaded_into_internal_db(ds_ct):+    await ds_ct.invoke_startup()+    ct_map = await ds_ct.get_column_types("data", "posts")+    # "markdown" is not a registered type, so it won't appear+    assert "body" not in ct_map+    assert ct_map["author_email"].name == "email"+    assert ct_map["author_email"].config is None+    assert ct_map["website"].name == "url"+    assert ct_map["metadata"].name == "json"+ + ⋯+@pytest.mark.asyncio+async def test_config_with_type_and_config(tmp_path_factory):+    class PointColumnType(ColumnType):+        name = "point"+        description = "Geographic point"+ +    class _Plugin:+        @hookimpl+        def register_column_types(self, datasette):+            return [PointColumnType]+ +    plugin = _Plugin()+    pm.register(plugin, name="test_point_ct")+    try:+        db_directory = tmp_path_factory.mktemp("dbs")+        db_path = str(db_directory / "data.db")+        db = sqlite3.connect(str(db_path))+        db.execute("vacuum")+        db.execute("create table geo (id integer primary key, location text)")+        ds = Datasette(+            [db_path],+            config={+                "databases": {+                    "data": {+                        "tables": {+                            "geo": {+                                "column_types": {+                                    "location": {+                                        "type": "point",+                                        "config": {"srid": 4326},+                                    }+                                }+                            }+                        }+                    }+                }+            },+        )+        await ds.invoke_startup()+        ct = await ds.get_column_type("data", "geo", "location")+        assert ct.name == "point"+        assert ct.config == {"srid": 4326}+        db.close()+        for database in ds.databases.values():+            if not database.is_memory:+                database.close()+    finally:+        pm.unregister(plugin, name="test_point_ct")+ + 

+# --- Datasette API methods ---+ + ⋯+@pytest.mark.asyncio+async def test_get_column_type(ds_ct):+    await ds_ct.invoke_startup()+    ct = await ds_ct.get_column_type("data", "posts", "author_email")+    assert isinstance(ct, ColumnType)+    assert ct.name == "email"+    assert ct.config is None+ + ⋯+@pytest.mark.asyncio+async def test_get_column_type_missing(ds_ct):+    await ds_ct.invoke_startup()+    ct = await ds_ct.get_column_type("data", "posts", "title")+    assert ct is None+ + ⋯+@pytest.mark.asyncio+async def test_set_and_remove_column_type(ds_ct):+    await ds_ct.invoke_startup()+    await ds_ct.set_column_type("data", "posts", "title", "email")+    ct = await ds_ct.get_column_type("data", "posts", "title")+    assert ct.name == "email"+    assert ct.config is None+ +    await ds_ct.remove_column_type("data", "posts", "title")+    ct = await ds_ct.get_column_type("data", "posts", "title")+    assert ct is None+ + ⋯+@pytest.mark.asyncio+async def test_set_column_type_with_config(ds_ct):+    await ds_ct.invoke_startup()+    await ds_ct.set_column_type("data", "posts", "title", "url", {"max_length": 200})+    ct = await ds_ct.get_column_type("data", "posts", "title")+    assert ct.name == "url"+    assert ct.config == {"max_length": 200}+ + 

+# --- Plugin registration ---+ + ⋯+@pytest.mark.asyncio+async def test_builtin_column_types_registered(ds_ct):+    """register_column_types returns classes; _column_types stores them by name."""+    await ds_ct.invoke_startup()+    assert "url" in ds_ct._column_types+    assert "email" in ds_ct._column_types+    assert "json" in ds_ct._column_types+    assert "nonexistent" not in ds_ct._column_types+ + ⋯+@pytest.mark.asyncio+async def test_column_type_class_attributes(ds_ct):+    await ds_ct.invoke_startup()+    url_cls = ds_ct._column_types["url"]+    assert url_cls.name == "url"+    assert url_cls.description == "URL"+    email_cls = ds_ct._column_types["email"]+    assert email_cls.name == "email"+    assert email_cls.description == "Email address"+ + 

+# --- JSON API ---+ + ⋯+@pytest.mark.asyncio+async def test_column_types_extra(ds_ct):+    await ds_ct.invoke_startup()+    response = await ds_ct.client.get("/data/posts.json?_extra=column_types")+    assert response.status_code == 200+    data = response.json()+    assert "column_types" in data+    assert data["column_types"]["author_email"] == {"type": "email", "config": None}+    assert data["column_types"]["website"] == {"type": "url", "config": None}+    assert data["column_types"]["metadata"] == {"type": "json", "config": None}+    # "markdown" is not a registered type, so body should not appear+    assert "body" not in data["column_types"]+    # title has no column type, should not appear+    assert "title" not in data["column_types"]+ + ⋯+@pytest.mark.asyncio+async def test_display_columns_include_column_type(ds_ct):+    await ds_ct.invoke_startup()+    response = await ds_ct.client.get("/data/posts.json?_extra=display_columns")+    assert response.status_code == 200+    data = response.json()+    cols = {c["name"]: c for c in data["display_columns"]}+    assert cols["author_email"]["column_type"] == "email"+    assert cols["author_email"]["column_type_config"] is None+    assert cols["website"]["column_type"] == "url"+    assert cols["title"]["column_type"] is None+ + 

+# --- Rendering ---+ + ⋯+@pytest.mark.asyncio+async def test_url_render_cell(ds_ct):+    await ds_ct.invoke_startup()+    response = await ds_ct.client.get("/data/posts.json?_extra=render_cell")+    assert response.status_code == 200+    data = response.json()+    rendered = data["render_cell"][0]+    assert "href" in rendered["website"]+    assert "https://example.com" in rendered["website"]+ + ⋯+@pytest.mark.asyncio+async def test_email_render_cell(ds_ct):+    await ds_ct.invoke_startup()+    response = await ds_ct.client.get("/data/posts.json?_extra=render_cell")+    assert response.status_code == 200+    data = response.json()+    rendered = data["render_cell"][0]+    assert "mailto:" in rendered["author_email"]+    assert "test@example.com" in rendered["author_email"]+ + ⋯+@pytest.mark.asyncio+async def test_json_render_cell(ds_ct):+    await ds_ct.invoke_startup()+    response = await ds_ct.client.get("/data/posts.json?_extra=render_cell")+    assert response.status_code == 200+    data = response.json()+    rendered = data["render_cell"][0]+    assert "<pre>" in rendered["metadata"]+ + 

+# --- Validation ---+ + ⋯+@pytest.mark.asyncio+async def test_email_validation_on_insert(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/-/insert",+        json={"row": {"title": "Test", "author_email": "not-an-email"}},+        headers=_headers(token),+    )+    assert response.status_code == 400+    assert "author_email" in response.json()["errors"][0]+ + ⋯+@pytest.mark.asyncio+async def test_email_validation_passes_valid(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/-/insert",+        json={"row": {"title": "Test", "author_email": "valid@example.com"}},+        headers=_headers(token),+    )+    assert response.status_code == 201+ + ⋯+@pytest.mark.asyncio+async def test_url_validation_on_insert(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/-/insert",+        json={"row": {"title": "Test", "website": "not-a-url"}},+        headers=_headers(token),+    )+    assert response.status_code == 400+    assert "website" in response.json()["errors"][0]+ + ⋯+@pytest.mark.asyncio+async def test_json_validation_on_insert(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/-/insert",+        json={"row": {"title": "Test", "metadata": "not-json{"}},+        headers=_headers(token),+    )+    assert response.status_code == 400+    assert "metadata" in response.json()["errors"][0]+ + ⋯+@pytest.mark.asyncio+async def test_validation_on_update(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/1/-/update",+        json={"update": {"author_email": "invalid"}},+        headers=_headers(token),+    )+    assert response.status_code == 400+    assert "author_email" in response.json()["errors"][0]+ + ⋯+@pytest.mark.asyncio+async def test_validation_allows_null(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/-/insert",+        json={"row": {"title": "Test", "author_email": None}},+        headers=_headers(token),+    )+    assert response.status_code == 201+ + ⋯+@pytest.mark.asyncio+async def test_validation_allows_empty_string(ds_ct):+    await ds_ct.invoke_startup()+    token = write_token(ds_ct)+    response = await ds_ct.client.post(+        "/data/posts/-/insert",+        json={"row": {"title": "Test", "author_email": ""}},+        headers=_headers(token),+    )+    assert response.status_code == 201+ + 

+# --- ColumnType base class ---+ + ⋯+@pytest.mark.asyncio+async def test_column_type_base_defaults():+    class TestType(ColumnType):+        name = "test"+        description = "Test type"+ +    ct = TestType()+    assert ct.config is None+    assert await ct.render_cell("val", "col", "tbl", "db", None, None) is None+    assert await ct.validate("val", None) is None+    assert await ct.transform_value("val", None) == "val"+ + 

+# --- render_cell extra with column types ---+ + ⋯+@pytest.mark.asyncio+async def test_render_cell_extra_with_column_types(ds_ct):+    await ds_ct.invoke_startup()+    response = await ds_ct.client.get("/data/posts.json?_extra=render_cell")+    assert response.status_code == 200+    data = response.json()+    rendered = data["render_cell"][0]+    assert "mailto:" in rendered["author_email"]+    assert "href" in rendered["website"]+ + ⋯+# --- Duplicate column type name ---+ + ⋯+@pytest.mark.asyncio+async def test_duplicate_column_type_name_raises_error():+    class DuplicateUrlType(ColumnType):+        name = "url"+        description = "Duplicate URL"+ +        async def render_cell(self, value, column, table, database, datasette, request):+            return None+ +    class _Plugin:+        @hookimpl+        def register_column_types(self, datasette):+            return [DuplicateUrlType]+ +    plugin = _Plugin()+    pm.register(plugin, name="test_duplicate_ct")+    try:+        ds = Datasette()+        with pytest.raises(StartupError, match="Duplicate column type name: url"):+            await ds.invoke_startup()+    finally:+        pm.unregister(plugin, name="test_duplicate_ct")+ +