Plugin System

The CountryFlag package includes a flexible plugin system that allows you to extend its functionality.

Overview

The plugin system allows you to:

  • Add custom data sources for country information

  • Implement custom caching mechanisms

  • Add new output formats

  • Customize flag rendering

Creating a Plugin

To create a plugin, you need to implement the BasePlugin interface:

from countryflag.plugins.base import BasePlugin
from countryflag.core.models import CountryInfo
from typing import List, Optional

class CustomPlugin(BasePlugin):
    def get_country_info(self, name: str) -> Optional[CountryInfo]:
        """Get country information for a given country name."""
        # Implementation here
        pass

    def get_supported_countries(self) -> List[CountryInfo]:
        """Get a list of supported countries."""
        # Implementation here
        pass

    def get_supported_regions(self) -> List[str]:
        """Get a list of supported regions/continents."""
        # Implementation here
        pass

    def get_countries_by_region(self, region: str) -> List[CountryInfo]:
        """Get countries in a specific region/continent."""
        # Implementation here
        pass

Example Plugins

Custom Data Source

Here's an example of a plugin that uses a JSON file as a data source:

custom_data_source_plugin.py
  1#!/usr/bin/env python3
  2"""
  3Example of a custom data source plugin for countryflag.
  4
  5This plugin uses a custom data source (a JSON file) for country information
  6instead of the default country_converter library.
  7"""
  8
  9import json
 10import os
 11from typing import Any, Dict, List, Optional
 12
 13from countryflag.core.models import CountryInfo
 14from countryflag.plugins.base import BasePlugin
 15
 16
 17class CustomDataSourcePlugin(BasePlugin):
 18    """
 19    Custom data source plugin for countryflag.
 20
 21    This plugin uses a JSON file as the data source for country information.
 22
 23    Attributes:
 24        data_file: Path to the JSON file containing country data.
 25        _countries: Dictionary mapping country names to country information.
 26        _iso2_map: Dictionary mapping ISO2 codes to country information.
 27        _flag_map: Dictionary mapping flag emojis to country information.
 28    """
 29
 30    def __init__(self, data_file: str):
 31        """
 32        Initialize the plugin with a custom data file.
 33
 34        Args:
 35            data_file: Path to the JSON file containing country data.
 36        """
 37        self.data_file = data_file
 38        self._countries = {}
 39        self._iso2_map = {}
 40        self._flag_map = {}
 41        self._regions = {}
 42
 43        # Load data from the file
 44        self._load_data()
 45
 46    def _load_data(self) -> None:
 47        """Load country data from the JSON file."""
 48        if not os.path.exists(self.data_file):
 49            raise FileNotFoundError(f"Data file not found: {self.data_file}")
 50
 51        with open(self.data_file, encoding="utf-8") as f:
 52            data = json.load(f)
 53
 54        # Process the data
 55        for item in data:
 56            country_info = CountryInfo(
 57                name=item["name"],
 58                iso2=item["iso2"],
 59                iso3=item.get("iso3", ""),
 60                official_name=item.get("official_name", item["name"]),
 61                region=item.get("region", ""),
 62                subregion=item.get("subregion", ""),
 63            )
 64
 65            # Add to the dictionaries
 66            self._countries[country_info.name.lower()] = country_info
 67            self._iso2_map[country_info.iso2.lower()] = country_info
 68            self._flag_map[country_info.flag] = country_info
 69
 70            # Add to regions
 71            if country_info.region:
 72                if country_info.region not in self._regions:
 73                    self._regions[country_info.region] = []
 74                self._regions[country_info.region].append(country_info)
 75
 76    def get_country_info(self, name: str) -> Optional[CountryInfo]:
 77        """
 78        Get country information for a given country name.
 79
 80        Args:
 81            name: The country name to look up.
 82
 83        Returns:
 84            CountryInfo: The country information, or None if the country is not found.
 85        """
 86        # Try direct lookup
 87        if name.lower() in self._countries:
 88            return self._countries[name.lower()]
 89
 90        # Try ISO2 lookup
 91        if len(name) == 2 and name.lower() in self._iso2_map:
 92            return self._iso2_map[name.lower()]
 93
 94        # Not found
 95        return None
 96
 97    def get_supported_countries(self) -> List[CountryInfo]:
 98        """
 99        Get a list of supported countries.
100
101        Returns:
102            List[CountryInfo]: A list of country information objects.
103        """
104        return list(self._countries.values())
105
106    def get_supported_regions(self) -> List[str]:
107        """
108        Get a list of supported regions/continents.
109
110        Returns:
111            List[str]: A list of supported regions/continents.
112        """
113        return list(self._regions.keys())
114
115    def get_countries_by_region(self, region: str) -> List[CountryInfo]:
116        """
117        Get countries in a specific region/continent.
118
119        Args:
120            region: The region/continent name.
121
122        Returns:
123            List[CountryInfo]: A list of countries in the specified region.
124        """
125        return self._regions.get(region, [])
126
127    def convert_country_name(self, name: str, to_format: str) -> str:
128        """
129        Convert a country name to the specified format.
130
131        Args:
132            name: The country name to convert.
133            to_format: The format to convert to (e.g., "ISO2", "ISO3").
134
135        Returns:
136            str: The converted country code, or "not found" if the country is not found.
137        """
138        country_info = self.get_country_info(name)
139        if not country_info:
140            return "not found"
141
142        if to_format == "ISO2":
143            return country_info.iso2
144        elif to_format == "ISO3":
145            return country_info.iso3
146        else:
147            return "not found"
148
149    def get_flag(self, country_name: str) -> Optional[str]:
150        """
151        Get the flag emoji for a country name.
152
153        Args:
154            country_name: The country name to get the flag for.
155
156        Returns:
157            str: The flag emoji, or None if the country is not found.
158        """
159        country_info = self.get_country_info(country_name)
160        return country_info.flag if country_info else None
161
162    def reverse_lookup(self, flag_emoji: str) -> Optional[str]:
163        """
164        Get the country name for a flag emoji.
165
166        Args:
167            flag_emoji: The flag emoji to look up.
168
169        Returns:
170            str: The country name, or None if the flag is not found.
171        """
172        if flag_emoji in self._flag_map:
173            return self._flag_map[flag_emoji].name
174        return None
175
176
177# Example data file (sample_countries.json)
178SAMPLE_DATA = [
179    {
180        "name": "United States",
181        "iso2": "US",
182        "iso3": "USA",
183        "official_name": "United States of America",
184        "region": "Americas",
185        "subregion": "Northern America",
186    },
187    {
188        "name": "Canada",
189        "iso2": "CA",
190        "iso3": "CAN",
191        "official_name": "Canada",
192        "region": "Americas",
193        "subregion": "Northern America",
194    },
195    {
196        "name": "Germany",
197        "iso2": "DE",
198        "iso3": "DEU",
199        "official_name": "Federal Republic of Germany",
200        "region": "Europe",
201        "subregion": "Western Europe",
202    },
203    {
204        "name": "France",
205        "iso2": "FR",
206        "iso3": "FRA",
207        "official_name": "French Republic",
208        "region": "Europe",
209        "subregion": "Western Europe",
210    },
211]
212
213
214def create_sample_data_file(output_file: str = "sample_countries.json") -> str:
215    """
216    Create a sample data file for the custom data source plugin.
217
218    Args:
219        output_file: Path to the output file.
220
221    Returns:
222        str: Path to the created file.
223    """
224    with open(output_file, "w", encoding="utf-8") as f:
225        json.dump(SAMPLE_DATA, f, indent=2)
226    return output_file
227
228
229def example_usage():
230    """Example usage of the custom data source plugin."""
231    # Create a sample data file
232    data_file = create_sample_data_file()
233
234    # Create the plugin
235    plugin = CustomDataSourcePlugin(data_file)
236
237    # Register the plugin with countryflag
238    from countryflag.plugins import register_plugin
239
240    register_plugin("custom_data", plugin)
241
242    # Use the plugin with CountryFlag
243    from countryflag.core import CountryFlag
244
245    cf = CountryFlag()
246
247    # Get a flag
248    flags, pairs = cf.get_flag(["United States", "Germany"])
249    print("Flags:", flags)
250    print("Pairs:", pairs)
251
252    # Get countries by region
253    flags, pairs = cf.get_flags_by_region("Europe")
254    print("European flags:", flags)
255    print("European pairs:", pairs)
256
257    # Clean up
258    import os
259
260    os.remove(data_file)
261
262
263if __name__ == "__main__":
264    example_usage()

Custom Cache

Example of a Redis-based cache plugin:

custom_cache_plugin.py
  1#!/usr/bin/env python3
  2"""
  3Example of a custom cache plugin for countryflag.
  4
  5This plugin implements a Redis-based cache for countryflag.
  6"""
  7
  8import json
  9import pickle
 10from typing import Any, Optional
 11
 12try:
 13    import redis
 14except ImportError:
 15    redis = None
 16
 17from countryflag.cache.base import Cache
 18
 19
 20class RedisCache(Cache):
 21    """
 22    Redis-based cache implementation.
 23
 24    This class implements a cache that stores data in Redis.
 25
 26    Attributes:
 27        _redis: Redis client.
 28        _prefix: Prefix for Redis keys.
 29        _ttl: Time-to-live for cache entries (in seconds).
 30    """
 31
 32    def __init__(
 33        self,
 34        host: str = "localhost",
 35        port: int = 6379,
 36        db: int = 0,
 37        prefix: str = "countryflag:",
 38        ttl: int = 3600,
 39    ):
 40        """
 41        Initialize the Redis cache.
 42
 43        Args:
 44            host: Redis host.
 45            port: Redis port.
 46            db: Redis database number.
 47            prefix: Prefix for Redis keys.
 48            ttl: Time-to-live for cache entries (in seconds).
 49
 50        Raises:
 51            ImportError: If Redis is not installed.
 52        """
 53        if redis is None:
 54            raise ImportError(
 55                "Redis is not installed. Install it with 'pip install redis'."
 56            )
 57
 58        self._redis = redis.Redis(host=host, port=port, db=db)
 59        self._prefix = prefix
 60        self._ttl = ttl
 61
 62    def _get_key(self, key: str) -> str:
 63        """
 64        Get the full Redis key with prefix.
 65
 66        Args:
 67            key: The cache key.
 68
 69        Returns:
 70            str: The full Redis key.
 71        """
 72        return f"{self._prefix}{key}"
 73
 74    def get(self, key: str) -> Optional[Any]:
 75        """
 76        Get a value from the cache.
 77
 78        Args:
 79            key: The cache key.
 80
 81        Returns:
 82            The cached value, or None if the key is not in the cache.
 83        """
 84        redis_key = self._get_key(key)
 85        value = self._redis.get(redis_key)
 86
 87        if value is None:
 88            return None
 89
 90        try:
 91            # Try to deserialize with pickle
 92            return pickle.loads(value)
 93        except (pickle.PickleError, TypeError, ValueError):
 94            # Fall back to JSON
 95            try:
 96                return json.loads(value)
 97            except json.JSONDecodeError:
 98                # Return as string
 99                return value.decode("utf-8")
100
101    def set(self, key: str, value: Any) -> None:
102        """
103        Set a value in the cache.
104
105        Args:
106            key: The cache key.
107            value: The value to cache.
108        """
109        redis_key = self._get_key(key)
110
111        try:
112            # Try to serialize with pickle
113            serialized = pickle.dumps(value)
114        except (pickle.PickleError, TypeError):
115            # Fall back to JSON
116            try:
117                serialized = json.dumps(value).encode("utf-8")
118            except (TypeError, ValueError):
119                # Fall back to string
120                serialized = str(value).encode("utf-8")
121
122        self._redis.set(redis_key, serialized, ex=self._ttl)
123
124    def delete(self, key: str) -> None:
125        """
126        Delete a value from the cache.
127
128        Args:
129            key: The cache key to delete.
130        """
131        redis_key = self._get_key(key)
132        self._redis.delete(redis_key)
133
134    def clear(self) -> None:
135        """
136        Clear all values from the cache.
137        """
138        # Get all keys with the prefix
139        pattern = f"{self._prefix}*"
140        keys = self._redis.keys(pattern)
141
142        # Delete all keys
143        if keys:
144            self._redis.delete(*keys)
145
146    def contains(self, key: str) -> bool:
147        """
148        Check if a key exists in the cache.
149
150        Args:
151            key: The cache key to check.
152
153        Returns:
154            bool: True if the key exists in the cache, False otherwise.
155        """
156        redis_key = self._get_key(key)
157        return self._redis.exists(redis_key) > 0
158
159
160def example_usage():
161    """Example usage of the Redis cache plugin."""
162    if redis is None:
163        print("Redis is not installed. Install it with 'pip install redis'.")
164        return
165
166    try:
167        # Create the Redis cache
168        redis_cache = RedisCache()
169
170        # Use the cache with CountryFlag
171        from countryflag.core import CountryFlag
172
173        cf = CountryFlag(cache=redis_cache)
174
175        # First request (cache miss)
176        import time
177
178        start_time = time.time()
179        flags1, _ = cf.get_flag(["United States", "Canada", "Germany"])
180        miss_time = time.time() - start_time
181
182        # Second request (cache hit)
183        start_time = time.time()
184        flags2, _ = cf.get_flag(["United States", "Canada", "Germany"])
185        hit_time = time.time() - start_time
186
187        # Print results
188        print("Flags:", flags1)
189        print(f"Cache miss time: {miss_time:.6f} seconds")
190        print(f"Cache hit time: {hit_time:.6f} seconds")
191        print(f"Speed improvement: {miss_time / hit_time:.2f}x")
192
193        # Clean up
194        redis_cache.clear()
195
196    except redis.ConnectionError:
197        print("Could not connect to Redis. Make sure Redis is running.")
198
199
200if __name__ == "__main__":
201    example_usage()

Custom Output Format

Example of a plugin that adds HTML and XML output formats:

custom_output_format_plugin.py
  1#!/usr/bin/env python3
  2"""
  3Example of a custom output format plugin for countryflag.
  4
  5This plugin adds support for HTML and XML output formats.
  6"""
  7
  8import xml.dom.minidom
  9import xml.etree.ElementTree as ET
 10from typing import Any, Callable, Dict, List, Tuple
 11
 12from countryflag.core import CountryFlag
 13
 14
 15class OutputFormatPlugin:
 16    """
 17    Plugin for custom output formats.
 18
 19    This class adds support for HTML and XML output formats.
 20    """
 21
 22    def __init__(self):
 23        """Initialize the plugin."""
 24        self._original_format_output = None
 25
 26    def _format_as_html(self, pairs: List[Tuple[str, str]]) -> str:
 27        """
 28        Format the output as HTML.
 29
 30        Args:
 31            pairs: A list of (country, flag) pairs.
 32
 33        Returns:
 34            str: The HTML output.
 35        """
 36        html = [
 37            "<!DOCTYPE html>",
 38            "<html>",
 39            "<head>",
 40            '    <meta charset="UTF-8">',
 41            "    <title>Country Flags</title>",
 42            "    <style>",
 43            "        table { border-collapse: collapse; width: 100%; }",
 44            "        th, td { text-align: left; padding: 8px; }",
 45            "        tr:nth-child(even) { background-color: #f2f2f2; }",
 46            "        th { background-color: #4CAF50; color: white; }",
 47            "    </style>",
 48            "</head>",
 49            "<body>",
 50            "    <h1>Country Flags</h1>",
 51            "    <table>",
 52            "        <tr>",
 53            "            <th>Country</th>",
 54            "            <th>Flag</th>",
 55            "        </tr>",
 56        ]
 57
 58        for country, flag in pairs:
 59            html.append("        <tr>")
 60            html.append(f"            <td>{country}</td>")
 61            html.append(f"            <td>{flag}</td>")
 62            html.append("        </tr>")
 63
 64        html.extend(["    </table>", "</body>", "</html>"])
 65
 66        return "\n".join(html)
 67
 68    def _format_as_xml(self, pairs: List[Tuple[str, str]]) -> str:
 69        """
 70        Format the output as XML.
 71
 72        Args:
 73            pairs: A list of (country, flag) pairs.
 74
 75        Returns:
 76            str: The XML output.
 77        """
 78        root = ET.Element("countries")
 79
 80        for country, flag in pairs:
 81            country_elem = ET.SubElement(root, "country")
 82            name_elem = ET.SubElement(country_elem, "name")
 83            name_elem.text = country
 84            flag_elem = ET.SubElement(country_elem, "flag")
 85            flag_elem.text = flag
 86
 87        # Convert to string with pretty formatting
 88        xml_str = ET.tostring(root, encoding="unicode")
 89        dom = xml.dom.minidom.parseString(xml_str)
 90        return dom.toprettyxml(indent="    ")
 91
 92    def patch_format_output(self, country_flag: CountryFlag) -> None:
 93        """
 94        Patch the format_output method of CountryFlag to add support for HTML and XML.
 95
 96        Args:
 97            country_flag: The CountryFlag instance to patch.
 98        """
 99        # Save the original method
100        self._original_format_output = country_flag.format_output
101
102        # Define the new method
103        def new_format_output(
104            pairs: List[Tuple[str, str]],
105            output_format: str = "text",
106            separator: str = " ",
107        ) -> str:
108            if output_format == "html":
109                return self._format_as_html(pairs)
110            elif output_format == "xml":
111                return self._format_as_xml(pairs)
112            else:
113                # Call the original method
114                return self._original_format_output(pairs, output_format, separator)
115
116        # Replace the method
117        country_flag.format_output = new_format_output.__get__(
118            country_flag, CountryFlag
119        )
120
121    def restore_format_output(self, country_flag: CountryFlag) -> None:
122        """
123        Restore the original format_output method.
124
125        Args:
126            country_flag: The CountryFlag instance to restore.
127        """
128        if self._original_format_output:
129            country_flag.format_output = self._original_format_output
130
131
132def example_usage():
133    """Example usage of the output format plugin."""
134    # Create the CountryFlag instance
135    from countryflag.core import CountryFlag
136
137    cf = CountryFlag()
138
139    # Create and apply the plugin
140    plugin = OutputFormatPlugin()
141    plugin.patch_format_output(cf)
142
143    # Convert some country names to flags
144    _, pairs = cf.get_flag(["United States", "Canada", "Germany"])
145
146    # Format the output in different formats
147    html_output = cf.format_output(pairs, output_format="html")
148    xml_output = cf.format_output(pairs, output_format="xml")
149
150    # Print results
151    print("HTML Output:")
152    print(html_output[:500] + "...\n")  # Show first 500 chars
153
154    print("XML Output:")
155    print(xml_output)
156
157    # Restore the original method
158    plugin.restore_format_output(cf)
159
160
161if __name__ == "__main__":
162    example_usage()

Using Plugins

To use a plugin:

from countryflag.plugins import register_plugin
from countryflag.core import CountryFlag
from my_plugin import CustomPlugin

# Create and register the plugin
plugin = CustomPlugin()
register_plugin("custom_plugin", plugin)

# Use the plugin
cf = CountryFlag()
flags = cf.get_flag(["United States", "Canada"])

Plugin API Reference

class countryflag.plugins.base.BasePlugin[source]

Bases: ABC

Abstract base class for plugins.

This class defines the interface that all plugins must adhere to.

abstract get_country_info(name)[source]

Get country information for a given country name.

Parameters:

name (str) – The country name to look up.

Returns:

The country information, or None if the country is not found.

Return type:

CountryInfo

abstract get_supported_countries()[source]

Get a list of supported countries.

Returns:

A list of country information objects.

Return type:

List[CountryInfo]

abstract get_supported_regions()[source]

Get a list of supported regions/continents.

Returns:

A list of supported regions/continents.

Return type:

List[str]

abstract get_countries_by_region(region)[source]

Get countries in a specific region/continent.

Parameters:

region (str) – The region/continent name.

Returns:

A list of countries in the specified region.

Return type:

List[CountryInfo]

abstract convert_country_name(name, to_format)[source]

Convert a country name to the specified format.

Parameters:

  • name (str) – The country name to convert.

  • to_format (str) – The format to convert to (e.g., “ISO2”, “ISO3”).

Returns:

The converted country code, or “not found” if the country is not found.

Return type:

str

abstract get_flag(country_name)[source]

Get the flag emoji for a country name.

Parameters:

country_name (str) – The country name to get the flag for.

Returns:

The flag emoji, or None if the country is not found.

Return type:

str

abstract reverse_lookup(flag_emoji)[source]

Get the country name for a flag emoji.

Parameters:

flag_emoji (str) – The flag emoji to look up.

Returns:

The country name, or None if the flag is not found.

Return type:

str

Best Practices

  1. Error Handling: Always implement proper error handling in your plugins

  2. Performance: Consider caching results for better performance

  3. Documentation: Document your plugin's behavior and requirements

  4. Testing: Write comprehensive tests for your plugin