bq-schema-policy-tags 0.7.3
Define your BigQuery tables as dataclasses. With this forked library you can also add Policy Tags to your dataclasses to protect your PII data.
Requires Python: >=3.7

# bq-schema-policy-tags
[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
[![Imports: isort](https://img.shields.io/badge/%20imports-isort-%231674b1?style=flat&labelColor=ef8336)](https://pycqa.github.io/isort/)

# This Project is a Fork of [bq-schema by Limehome GmbH](https://pypi.org/project/bq-schema/)

## Motivation (Original GH Repo)

At limehome we are heavy users of python and bigquery. This library was created to mainly solve the following issues:

* Define table schemas in code and have a migration script to apply changes.
* On deploy make sure that all schemas were applied, otherwise abort.
* Guarantee that when we try to write data to a table, the data matches the schema of the table (required / optional, datatypes)
* Version our tables and enable migrations to a new schema

Additionally this library aims to help the users through the usage of [python typing](https://docs.python.org/3/library/typing.html).

* Specify your schema as a [python dataclass](https://docs.python.org/3/library/dataclasses.html)
* Our migration script converts the data class into a bigquery schema definition
* Deserialize rows into a dataclass instance, while reading from a table
* Serialize a dataclass instance into a dictionary and write it to the table.

The main benefit of combining all these features is, that we can guarantee that our code will run, before we deploy to production.

## Motivation for Fork

The primary reason for forking this project was to extend its existing capabilities to support Policy Tags and Taxonomy tags in Google Cloud Platform (GCP). The original codebase provided a strong foundation for working with Python dataclasses, but it lacked features for adding metadata tags that can be crucial for managing sensitive or personally identifiable information (PII).

### Why Policy Tags?

Policy Tags allow for more granular control over data access. When dealing with sensitive or regulated data, such as PII, it's crucial to have mechanisms that enable fine-tuned permissions. Policy Tags in GCP make it possible to associate specific security and access policies with individual data fields, rather than entire data sets, thereby providing a more secure and manageable environment.

### Taxonomy Tags in GCP

Taxonomy Tags in GCP serve as a method for categorizing data based on its type or nature. By attaching taxonomy tags to data fields, organizations can more easily track, manage, and govern their data. This is particularly useful for compliance with data protection laws and regulations, as well as for internal data governance strategies.

By incorporating these features into Python dataclasses through this fork, we aim to provide a more robust, secure, and compliant data handling solution.


## Quickstart
Since this library makes use of newer features of python, you need at least python3.7.

1. Install the package
```
pip install bq-schema-policy-tags
```

2. Create a schema and a table definition in my_table.py
```python
@dataclass
class Schema:
    string_field: str = field(metadata={"description": "This is a STRING field."})
    int_field: Optional[int]
    some_floats: List[float]
    bool_field: bool

class MyTable(BigqueryTable):
    name = "my_table_name"
    schema = Schema
```

If you have already tables created in your account, you can use the convert-table script to create a schema.

*Note*: The script produces a file which is meant to be a starting point. You will most likely have to add some imports yourself!


3. Create your table

**Hint:** 
Make sure to have you credentials set:
```
export GOOGLE_APPLICATION_CREDENTIALS=your_auth.json
```

Alternativly you can set the service_file as a environment variable:
```
export GOOGLE_SERVICE_FILE={"type": "service_account", ...}
```

Now create your table
```
migrate-tables --project my_project --dataset my_dataset --module-path my_table --apply
```

4. Write a row
```python
from google.cloud import bigquery
row = Schema(string_field="foo", int_field=1, some_floats=[1.0, 2.0], bool_field=True)
row_transformer = RowTransformer[Schema](Schema)
serialized_row = RowTransformer.dataclass_instance_to_bq_row(row)

bigquery_client = bigquery.Client()
table = bigquery_client.get_table("project.dataset.my_table_name")
bigquery_client.insert_rows(table, [serialized_row])
```

5. Validate you code with a type checker like [mypy](http://mypy-lang.org/)
```
mypy my_table.py
```

6. Read a row
```python
query = "SELECT * FROM project.dataset.my_table_name"
for row in bigquery_client.query(query=query):
    deserialized_row = row_transformer.bq_row_to_dataclass_instance(row)
    assert isinstance(deserialized_row, Schema)
```


## Documentation

### Schema definitions
For a full list of supported types check the following schema:
```python
from typing import Optional
from dataclasses import dataclass
from bq_schema_policy_tags.types.type_mapping import Timestamp, Geography

@dataclass
class RequiredNestedField:
    int_field: int = field(metadata={"description": "This field is an INT field."})


@dataclass
class RequiredSchema:
    string_field: str = field(metadata={"description": "This field is a STRING field."})
    string_field_optional = Optional[str]
    bytes_field: bytes
    int_field: int
    float_field: float
    numeric_field: Decimal
    bool_field: bool
    timestamp_field: Timestamp
    date_field: date
    time_field: time
    datetime_field: datetime
    geography_field: Geography
    required_nested_field: RequiredNestedField = field(metadata={"description": "This field is a STRUCT field."})
    optional_nested_field: Optional[RequiredNestedField] 
    repeated_nested_field: List[RequiredNestedField]
```

### Adding Policy Tags (New)
To add policy tags to a field, include policy_tags in the field metadata.

```python
from dataclasses import dataclass, field

@dataclass
class ExampleWithPolicyTags:
    sensitive_data: str = field(metadata={"policy_tags": ["sensitive"]})

schema = dataclass_to_schema(ExampleWithPolicyTags)
```

#### Abstract tables

If you want to have an class that inherits from BigqueryTable but does not actually map to a table in BigQuery (is abstract, common interface etc.), you can have it inherit from ABC and it will not be discovered if you pass the flag "--ignore-abstract"

``` migrate-tables ... --ignore-abstract ```


```python
class SomeInterface(BigqueryTable, ABC):
    pass 
class ConcreteImplementation(SomeInterface):
    name="Some value here"
    schema = SomeSchema
```
Without the flag, this would fail due to name and schema being required in "SomeInterface"
#### Timestamps
Timestamps are deserialized into datetime objects, due to the nature of the underlying bq library. To distinguish between datetime and timestamp use bq_schema_policy_tags.types.type_mapping.
Usage:

```python
from bq_schema_policy_tags.types.type_mapping import Timestamp
from datetime import datetime

the_timestamp = Timestamp(datetime.utcnow())
```


#### Geography
This library treats the geography data type as a string. BigQuery accepts geography values either in the [WKT](https://en.wikipedia.org/wiki/Well-known_text_representation_of_geometry) or [GeoJson](https://geojson.org) format. To actually parse and work with geodata in python, one could use the [shapely](https://pypi.org/project/Shapely/) library. Here is an example how to load a point from the WKT format:
```python
from shapely.wkt import loads
loads('POINT (0 0)')
```

### Table definitions
The bigquery class is used for:
* Recursive table discovery by our migrate-tables script
* Define table properties like name and schema


### Required properties
* name: The name of the table
* schema: table schema either as dataclass or a list of schema fields

### Optional properties
* project: name of the project, can be overwritten by the migrate-tables script
* dataset: name of the dataset, can be overwritten by the migrate-tables script

#### Versioning tables
Since bigquery does not allow backwards incompatible schema changes, you might want to version your schemas.project
```python
class MyTable(BigqueryTable):
    name = "my_table_name"
    schema = Schema
    version = "1"
```
By default the version will be appended to the table name, like so: my_table_name_v1. If you want to overwrite this behaviour,
you can implement the full_table_name method.

#### Time partitioning
Define time partitioning for your table:
```python
from bq_schema_policy_tags.types.type_mapping import Timestamp
from google.cloud.bigquery import TimePartitioning, TimePartitioningType

class MyTable:
    time_partitioning = TimePartitioning(
        type_=TimePartitioningType.DAY, field="some_column"
    )
```

### Scripts

#### migrate-tables
This script has two uses:
* Check if locally defined schemas are in sync with the schemas in bigquery
* If a difference is detected, we try to apply the changes

The script will find all defined tables recursivly for a given python module.

**Note**: If you have not defined your project and / or dataset in code, you will have to pass it as a parameter to the script.
Show the help:
```
migrate-tables --help
```

Check if tables are in sync. List all changes.
```
migrate-tables --module-path module/
```

If you want the script to fail on a change, add the validate flag. Useful for running inside your CI:
```
migrate-tables --module-path module/ --validate
```

Apply changes
```
migrate-tables --module-path src/jobs/ --apply
```


#### convert-table
If you already have tables created in bigquery, this script print the corresponding dataclass for you.

*Note*: The script produces a file which is meant to be a starting point. You will most likely have to add some imports yourself!

Show the help:
```
convert-table --help
```

Print a table:
```
convert-table --project project --dataset scraper --table-name table_name >> schema.py
```

### Development

#### Setting up your dev environment
1) Clone the project.
2) Navigate into the cloned project.
3) Create a virtual environment with **python version >=3.7**

    `pipenv --python PYTHON_VERSION`
    ```
    $ pipenv --python 3.7
    ```
    or

    `virtualenv -p /PATH_TO_PYTHON/ /DESIRED_PATH/VENV_NAME`
    ```
    $ virtualenv -p /usr/bin/python3.7 placeholder
    ```

4) Install flit via pip
    ```
    $ pip install flit
    ```
5) Install packages

    ```
    $ flit install --symlink
    ```

#### Code quality
Run all code quality checks:
```
inv check-all
```
##### Test
```
inv test
```
##### Lint
```
inv lint
```
##### Types
```
inv type-check
```
##### Code format
```
inv format-code
```

Validate code is correctly formatted:
```
inv check-code-format
```

## Known Issues

### MissingValueError in `test_row_transformer.py`

When running the pytest suite, a `MissingValueError` may occur, specifically indicating a missing value for the field "string_field". This issue appears to be inherent to the original codebase and may not have a straightforward solution due to the project being no longer maintained.

#### Potential Causes

1. The `row` object in the test case may not contain the "string_field" and its value.
2. The `Schema` data class might be expecting a value for "string_field" which is not provided.
3. The function `dict_to_row(row_as_dict: dict) -> Row` may not be converting the dictionary to a `Row` object correctly.

#### Suggested Workarounds

1. Double-check the test data for completeness.
2. If possible, modify the `Schema` data class to provide a default value for "string_field".
3. Check the `RowTransformer` class and the `from_dict` function for issues related to field handling.

Please note that these are suggested workarounds and may not resolve the issue entirely.