イギリス・リバプール大学(University of Liverpool)の研究チームは、因果集合理論（causal set theory）と呼ばれる量子重力の新しい理論を使い、宇宙の始まりについて計算したところ、宇宙に始まりはなく無限の過去に常に存在していたという結果を得ました。
この結果に従うと、ビッグバンは宇宙が遂げた最近の進化の1つでしかないということになります。
この研究は『arXiv』で公開されたのみで、査読付き論文誌に掲載されてはいませんが、ビッグバン以前の宇宙は無だったという説明に納得できない人には、興味深い洞察になるかもしれません。
この研究成果は、2021年9月24日にプレプリントサーバー『arXiv』で公開されています。
⇧ また、証明ができ無さそうな理論ですな...
今が、2024年8月10日（土）だから、およそ、3年が経とうとしているわけなんだが、研究は続けられているんだろうか？
SQLAlchemyで主キー（primary key）の存在しないテーブルは扱えない？
公式のドキュメントにて、
How do I map a table that has no primary key?
The SQLAlchemy ORM, in order to map to a particular table, needs there to be at least one column denoted as a primary key column; multiple-column, i.e. composite, primary keys are of course entirely feasible as well. These columns do not need to be actually known to the database as primary key columns, though it’s a good idea that they are. It’s only necessary that the columns behave as a primary key does, e.g. as a unique and not nullable identifier for a row.
Most ORMs require that objects have some kind of primary key defined because the object in memory must correspond to a uniquely identifiable row in the database table; at the very least, this allows the object can be targeted for UPDATE and DELETE statements which will affect only that object’s row and no other. However, the importance of the primary key goes far beyond that. In SQLAlchemy, all ORM-mapped objects are at all times linked uniquely within a Session to their specific database row using a pattern called the identity map, a pattern that’s central to the unit of work system employed by SQLAlchemy, and is also key to the most common (and not-so-common) patterns of ORM usage.
⇧ とあるので、
『「SQLAlchemy」で何らかの処理をさせたいのであれば、テーブルに「主キー（primary key）」は必須らしい』
、と読み取れてしまうような説明に見えてしまうのだが...
う～む、ドキュメントをどう解釈したものか、難しいんだが...
何で、こんな話を気にしたのかというと、PostgreSQLの拡張機能として導入できる「TimescaleDB」というライブラリがあるんですが、
⇧ 上記サイト様によりますと、「主キー（primary key）」の存在しないテーブルを扱っていて、「TimescaleDB」の公式のドキュメントでも、
⇧「主キー（primary key）」の無いテーブルがサンプルになっておるのよ...
「サロゲートキー」とか付与してしまえば良いように思うんだが、「主キー（primary key）」の存在しないテーブルにおいては、「SQLAlchemy」の「ORM（Object Relational Mapping）」の機能は無力であると...
「RDBMS（Relational DataBase Management Systems）」以外を利用するような環境だと、「主キー（primary key）」の存在しないテーブルを扱うケースが多いのかね？
所謂、ビッグデータ的な領域における、テーブル構成は、全くもって想像を絶する世界であるからして、何も分からん...
SQLAlchemyで主キー（primary key）の存在しないテーブルを扱えないわけでは無いらしいが...
う～む、結局のところ、Pythonには、Javaの「JDBC（Java DataBase Conectivity）」的な、標準で利用できるAPIが用意されていないので、
⇧ 利用するPostgreSQLのドライバー毎の処理を記述せねばならんのか？
と思っていたら、
公式マニュアル（How do I map a table that has no primary key?）によれば、必ずテーブルに主キーの指定が必要で、これがないとマッピングできないそうです。
ですが、DWHのテーブルだと必ずしも主キーが明示的に設定されているとは限りません（候補キーがあるにしても）。困りましたね…。
ですが、回避策としては2つあります。
⇧ 上記サイト様によりますと、
解決策２：ORMを使わない
ORMだと必ず主キーが必要なので、そもそもORMを使わなければ回避できます。
要するに、Core (SQL Expression Language)を使用する方法です。
⇧ とあり、「SQLAlchemy」の機能の内の1つ「ORM（Object Relational Mapping）」の機能を利用しないのであれば、「主キー（primary key）」の存在しないテーブルに対して、SQL文を実行できるそうな。
つまり、
⇧「SQLAlchemy Core」に該当する機能を利用すれば、「主キー（primary key）」の存在しないテーブルに対応できるらしい。
というか、「SQLAlchemy」の公式のドキュメントの『How do I map a table that has no primary key?』のところで、「主キー（primary key）」の存在しないテーブルについても「SQLAlchemy」の機能で、SQL文の処理が可能である点について触れてくれていれば良いのに、何故か、言及してくれていないんよね...
う～む、謎過ぎる...
SQLAlchemyで主キーの存在しないテーブルを扱えないわけでは無いらしいが...実現方法が分かり辛い
で、「主キー（primary key）」の存在しないテーブルで、「SQLAlchemy」の機能を利用できるのかどうかを試行錯誤することになり、かなり時間を奪われましたが、結果として、
『「主キー（primary key）」の存在しないテーブルでも「SQLAlchemy」の機能でSQL文の実行は可能』
ということが確認できましたので備忘録として。
環境は、
⇧ 上記の記事の時のものを利用していきます。
今回、動作検証するための対象のPostgreSQLのテーブルについて、
⇧「主キー（primary key）」が存在しないことを確認。
「app」フォルダ配下が、Pythonでの処理を実装したものを配置していく感じ。（Pythonド素人なので、ディレクトリ構造は、JavaのMavenやGradleといったビルドツールでプロジェクトを作った時のディレクトリ構造を真似ています。）
「__pycache__」ディレクトリなどは無視していただければと。
C:\Users\Toshinobu\Desktop\soft_work\python_work\fastapi.
│ .coverage
│ .env
│ .sqlfluff
│ coverage.xml
│
├─.venv
│ │ pyvenv.cfg
│ │
│ ├─Include
│ │ └─site
│ │ └─python3.12
│ │ └─greenlet
│ │ greenlet.h
│ │
│ ├─Lib
│ │ └─site-packages
│ │ │ appdirs.py
│ │
│ │ ...省略
│ │
│ └─Scripts
│ activate
│ activate.bat
│ Activate.ps1
│ chardetect.exe
│ coverage-3.12.exe
│ coverage.exe
│ coverage3.exe
│ deactivate.bat
│ diff-cover.exe
│ diff-quality.exe
│ dotenv.exe
│ email_validator.exe
│ fastapi.exe
│ httpx.exe
│ markdown-it.exe
│ pip.exe
│ pip3.12.exe
│ pip3.exe
│ py.test.exe
│ pygmentize.exe
│ pytest.exe
│ python.exe
│ pythonw.exe
│ sqlfluff.exe
│ tqdm.exe
│ typer.exe
│ uvicorn.exe
│ watchfiles.exe
│
├─.vscode
│ launch.json
│ settings.json
│
└─app
├─src
│ ├─main
│ │ ├─py
│ │ │ │ main.py
│ │ │ │
│ │ │ ├─call
│ │ │ │ └─procedural_language
│ │ │ │ select_all_user_func.py
│ │ │ │
│ │ │ ├─code
│ │ │ │ │ gender.py
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ gendar.cpython-312.pyc
│ │ │ │ gender.cpython-312.pyc
│ │ │ │
│ │ │ ├─config
│ │ │ │ │ app_config.py
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ app_config.cpython-312.pyc
│ │ │ │
│ │ │ ├─controller
│ │ │ │ │ user_controller.py
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ user_controller.cpython-312.pyc
│ │ │ │
│ │ │ ├─core
│ │ │ │ │ dependency_inject.py
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ dependency_inject.cpython-312.pyc
│ │ │ │
│ │ │ ├─database
│ │ │ │ │ database.py
│ │ │ │ │ database_config_interface.py
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ database.cpython-312.pyc
│ │ │ │ database_config_interface.cpython-312.pyc
│ │ │ │
│ │ │ ├─entity
│ │ │ │ │ address_entity.py
│ │ │ │ │ user_entity.py
│ │ │ │ │
│ │ │ │ ├─base
│ │ │ │ │ │ declarative_base.py
│ │ │ │ │ │
│ │ │ │ │ └─__pycache__
│ │ │ │ │ declarative_base.cpython-312.pyc
│ │ │ │ │
│ │ │ │ ├─timescaledb
│ │ │ │ │ └─hyper_table
│ │ │ │ │ │ condition_entity.py
│ │ │ │ │ │
│ │ │ │ │ └─__pycache__
│ │ │ │ │ condition_entity.cpython-312.pyc
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ address_entity.cpython-312.pyc
│ │ │ │ user_entity.cpython-312.pyc
│ │ │ │
│ │ │ ├─repository
│ │ │ │ │ user_repository.py
│ │ │ │ │
│ │ │ │ ├─timescaledb
│ │ │ │ │ └─hyper_table
│ │ │ │ │ │ condition_repository.py
│ │ │ │ │ │
│ │ │ │ │ └─__pycache__
│ │ │ │ │ condition_repository.cpython-312.pyc
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ user_repository.cpython-312.pyc
│ │ │ │
│ │ │ ├─service
│ │ │ │ │ user_service.py
│ │ │ │ │
│ │ │ │ ├─timescaledb
│ │ │ │ │ └─hyper_table
│ │ │ │ │ │ condition_service.py
│ │ │ │ │ │
│ │ │ │ │ └─__pycache__
│ │ │ │ │ condition_service.cpython-312.pyc
│ │ │ │ │
│ │ │ │ └─__pycache__
│ │ │ │ user_service.cpython-312.pyc
│ │ │ │
│ │ │ └─__pycache__
│ │ │ main.cpython-312.pyc
│ │ │
│ │ └─resources
│ │ application-dev.yml
│ │
│ └─test
│ ├─py
│ │ └─service
│ │ │ test_user_service.py
│ │ │
│ │ ├─timescaledb
│ │ │ └─hyper_table
│ │ │ │ test_condition.py
│ │ │ │
│ │ │ └─__pycache__
│ │ │ test_condition.cpython-312-pytest-8.3.2.pyc
│ │ │
│ │ └─__pycache__
│ │ test_user_service.cpython-312-pytest-8.3.2.pyc
│ │
│ └─resources
└─tool
├─py
│ └─database
│ │ generate_create_table-ft-SQLAlchemy_method.py
│ │ generate_create_table.py
│ │
│ └─output
│ ├─ddl
│ │ create_table-AddressEntity.sql
│ │ create_table-UserEntity.sql
│ │
│ └─dml
└─sql
├─dcl
├─ddl
│ create_func-select_all_user_func.sql
│
└─dml
ソースコードは以下のような感じ。
■C:\Users\Toshinobu\Desktop\soft_work\python_work\fastapi\app\src\main\py\entity\timescaledb\hyper_table\condition_entity.py
from datetime import date, datetime
# from sqlalchemy.ext.declarative import declarative_base
# from app.src.main.py.entity.base.declarative_base import Base
# from pydantic import BaseModel
from typing import ClassVar
from dataclasses import dataclass
# SQLテーブルのカラム設定用
from sqlalchemy import (
MetaData,
Table,
Column,
Integer,
Float,
String,
Date,
DateTime,
Boolean,
ForeignKey,
)
# class ConditionEntity(Base):
# __tablename__ = "conditions"
# __table_args__ = {
# "comment": "気温情報",
# "extend_existing": True,
# "implicit_returning": False,
# }
# time: datetime = Column(name="address_id", type_=DateTime, comment="時間")
# location: str = Column(name="location", type_=String, comment="場所")
# device: str = Column(name="device", type_=String, comment="デバイス")
# temperature: str = Column(name="temperature", type_=Float, comment="気温")
# humdity: str = Column(name="humdity", type_=Float, comment="湿度")
# @dataclass(frozen=True)
@dataclass
class ConditionEntity:
time: datetime
location: str
device: str
temperature: float
humidity: float
# def __init__(self, time, location, device, temperature, humidity):
# self.time = time
# self.location = location
# self.device = device
# self.temperature = temperature
# self.humidity = humidity
metadata: ClassVar[MetaData] = MetaData()
table: ClassVar[Table] = Table(
"conditions",
metadata,
Column("time", DateTime, comment="時間"),
Column("location", String, comment="場所"),
Column("device", String, comment="デバイス"),
Column("temperature", Float, comment="気温"),
Column("humidity", Float, comment="湿度"),
extend_existing=True,
implicit_returning=False,
comment="気温情報",
)
# def __post_init__(self):
# self.metadata = MetaData()
# self.table = Table(
# "conditions",
# self.metadata,
# Column("time", DateTime, comment="時間"),
# Column("location", String, comment="場所"),
# Column("device", String, comment="デバイス"),
# Column("temperature", Float, comment="気温"),
# Column("humidity", Float, comment="湿度"),
# extend_existing=True,
# implicit_returning=False,
# comment="気温情報",
# )
■C:\Users\Toshinobu\Desktop\soft_work\python_work\fastapi\app\src\main\py\repository\timescaledb\hyper_table\condition_repository.py
from injector import inject, singleton
from datetime import datetime
from sqlalchemy import (
MetaData,
Table,
Column,
Float,
String,
DateTime,
select,
insert,
update,
delete,
func,
)
from app.src.main.py.database.database import DatabaseConnection
from app.src.main.py.entity.timescaledb.hyper_table.condition_entity import (
ConditionEntity,
)
@singleton
class ConditionRepository:
@inject
def __init__(self, db: DatabaseConnection) -> None:
self.db = db
# self._initialize()
# def _initialize(self):
# self.meta = MetaData()
# self.meta.reflect(bind=self.db.get_async_engine())
# self.table_condition = Table(
# "conditions",
# self.meta,
# Column("time", DateTime, comment="時間"),
# Column("location", String, comment="場所"),
# Column("device", String, comment="デバイス"),
# Column("temperature", Float, comment="気温"),
# Column("humdity", Float, comment="湿度"),
# extend_existing=True,
# )
# self.table_condition = ConditionEntity
# Count
async def count(self) -> int:
"""
レコード件数を取得する
"""
async with self.db.get_db() as db_session:
result = await db_session.execute(
# select([func.count(self.table_condition.c.time)])
select(func.count(ConditionEntity.table.c.time))
)
count = result.scalar_one()
return count
# Delete
async def delete(self, condition_time: datetime) -> bool:
"""
気温情報を削除する
"""
async with self.db.get_db() as db_session:
result = await db_session.execute(
# delete(self.table_condition).where(
# self.table_condition.c.time == condition_time
# )
delete(ConditionEntity.table).where(
ConditionEntity.table.c.time == condition_time
)
)
await db_session.commit()
return result.rowcount > 0
# Select
async def find_by_id(self, condition_time: datetime):
"""
気温情報を検索する
"""
async with self.db.get_db() as db_session:
result = await db_session.execute(
# select([self.table_condition]).where(
# self.table_condition.c.time == condition_time
# )
select(ConditionEntity.table).where(
ConditionEntity.table.c.time == condition_time
)
)
row = result.fetchone()
if row:
return ConditionEntity(**row._mapping)
return None
async def find_latest_one(self):
"""
気温情報を検索する（最新の1件）
"""
async with self.db.get_db() as db_session:
result = await db_session.execute(
# select([self.table_condition])
# .order_by(self.table_condition.c.time.desc())
# .limit(1)
select(ConditionEntity.table)
.order_by(ConditionEntity.table.c.time.desc())
.limit(1)
)
row = result.fetchone()
if row:
return ConditionEntity(**row._mapping)
return None
# Insert
async def insert(self, condition) -> bool:
"""
気温情報を登録する
"""
async with self.db.get_db() as db_session:
# await db_session.execute(
# # insert(self.table_condition).values(
# insert(ConditionEntity.table).values(
# time=condition.time,
# location=condition.location,
# device=condition.device,
# temperature=condition.temperature,
# humidity=condition.humidity,
# )
# )
await db_session.execute(
insert(ConditionEntity.table).values(**condition.__dict__)
)
await db_session.commit()
return True
# Update
async def update(self, condition) -> bool:
"""
気温情報を更新する
"""
async with self.db.get_db() as db_session:
# result = await db_session.execute(
# # update(self.table_condition)
# # .where(self.table_condition.c.time == condition.time)
# update(ConditionEntity.table)
# .where(ConditionEntity.table.c.time == condition.time)
# .values(
# location=condition.location,
# device=condition.device,
# temperature=condition.temperature,
# humidity=condition.humidity,
# )
# )
condition_dict = condition.__dict__.copy()
condition_dict.pop("time", None) # 'time'キーを削除
result = await db_session.execute(
update(ConditionEntity.table)
.where(ConditionEntity.table.c.time == condition.time)
.values(**condition_dict)
)
await db_session.commit()
return result.rowcount > 0
■C:\Users\Toshinobu\Desktop\soft_work\python_work\fastapi\app\src\main\py\service\timescaledb\hyper_table\condition_service.py
from app.src.main.py.entity.timescaledb.hyper_table.condition_entity import (
ConditionEntity,
)
from app.src.main.py.repository.timescaledb.hyper_table.condition_repository import (
ConditionRepository,
)
from injector import inject, singleton
@singleton
class ConditionService:
@inject
def __init__(self, Condition_repository: ConditionRepository):
self.Condition_repository = Condition_repository
async def count(self) -> int:
return await self.Condition_repository.count()
async def delete(self, Condition_id: int) -> bool:
return await self.Condition_repository.delete(Condition_id)
async def find_condition(self, Condition_id: int) -> ConditionEntity:
return await self.Condition_repository.find_by_id(Condition_id)
async def find_latest_one(self) -> ConditionEntity:
return await self.Condition_repository.find_latest_one()
async def insert(self, Condition: ConditionEntity) -> bool:
return await self.Condition_repository.insert(Condition)
async def update(self, Condition: ConditionEntity) -> bool:
return await self.Condition_repository.update(Condition)
動確認用は、以下。
■C:\Users\Toshinobu\Desktop\soft_work\python_work\fastapi\app\src\test\py\service\timescaledb\hyper_table\test_condition.py
import pytest
import asyncio
import sys
from datetime import date, datetime, timedelta
from pprint import pprint
from injector import Injector, inject
# from app.src.main.py.config.app_config import AppConfig
from app.src.main.py.service.timescaledb.hyper_table.condition_service import (
ConditionService,
)
from app.src.main.py.entity.timescaledb.hyper_table.condition_entity import (
ConditionEntity,
)
# from app.src.main.py.repository.condition_repository import ConditionRepository
from app.src.main.py.core.dependency_inject import di_injector
from app.src.main.py.code.gender import Gender
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
class TestConditionService:
# def __del__(self) -> None:
# print("TestConditionServiceのデストラクタを実行")
# @pytest.fixture(autouse=True)
# def setup(self):
# self.condition_service = ConditionService # ConditionServiceのインスタンスを作成
# yield
# #@inject
# @pytest.fixture(autouse=True)
# def execute_before_test(self, condition_service: ConditionService):
# """
# テスト実行前処理
# """
# self.condition_service = condition_service
# yield
# @pytest.fixture(scope="class", autouse=True)
# def di(self):
# #DependencyInjector.get_class(DependencyInjector, AppConfig)
# # Injector([AppConfig()])
@pytest.fixture(scope="function", autouse=True)
def execute_before_test(self, request):
"""
テスト実行前処理
"""
print("\n")
print("【テスト開始】" + str(request.node.name))
yield
@pytest.fixture(scope="function", autouse=True)
def execute_after_test(self, request):
"""
テスト実行後処理
"""
yield
print("【テスト完了】" + str(request.node.name), end="")
# del(self)
### テスト ###
################################################################
# テスト01
# ・insert
################################################################
@pytest.mark.asyncio
async def test_01(self):
condition_service = di_injector.get_class(ConditionService)
condition_time: datetime = datetime.now() - timedelta(hours=1)
condition: ConditionEntity = ConditionEntity(
time=datetime.now(),
location="hamamatsu",
device="アナログ温湿度計",
temperature=float(41.3),
humidity=float(88),
)
condition_count: int = await condition_service.count()
print(condition_count)
if condition_count != 0:
# 最新の気温情報を取得
latest_condition: ConditionEntity = (
await condition_service.find_latest_one()
)
condition.time = condition_time
await condition_service.insert(condition)
################################################################
# テスト02
# ・count
# ・select
################################################################
@pytest.mark.asyncio
async def test_02(self):
condition_service = di_injector.get_class(ConditionService)
condition_time: datetime = datetime.now() - timedelta(hours=1)
condition_count: int = await condition_service.count()
print("■■■レコード件数■■■")
print(condition_count)
if condition_count != 0:
# 最新の気温情報を取得
latest_condition: ConditionEntity = (
await condition_service.find_latest_one()
)
condition_time = latest_condition.time
condition: ConditionEntity = await condition_service.find_condition(
condition_time
)
print("■■■取得結果■■■")
print(type(condition))
pprint(condition)
################################################################
# テスト03
# ・update
################################################################
@pytest.mark.asyncio
async def test_03(self):
condition_service = di_injector.get_class(ConditionService)
condition_time: datetime = datetime.now() - timedelta(hours=1)
condition_count: int = await condition_service.count()
print("■■■レコード件数■■■")
print(condition_count)
if condition_count != 0:
# 最新の気温情報を取得
latest_condition: ConditionEntity = (
await condition_service.find_latest_one()
)
condition_time = latest_condition.time
condition: ConditionEntity = await condition_service.find_condition(
condition_time
)
print("■■■取得結果■■■")
print(type(condition))
pprint(condition)
# 更新処理
condition.device = "XXXの温湿度計"
result: bool = await condition_service.update(condition)
print(result)
################################################################
# テスト04
# ・delete
################################################################
@pytest.mark.asyncio
async def test_04(self):
condition_service = di_injector.get_class(ConditionService)
condition_time: datetime = datetime.now() - timedelta(hours=1)
condition_count: int = await condition_service.count()
print("■■■レコード件数■■■")
print(condition_count)
if condition_count != 0:
# 最新の気温情報を取得
latest_condition: ConditionEntity = (
await condition_service.find_latest_one()
)
condition_time = latest_condition.time
result: bool = await condition_service.delete(condition_time)
print(result)
で、保存。
「pytest」で動作確認用の処理を実行。
一応、動作はしましたと。
ただ、Pythonにおける実装の基本形みたいなのがサッパリ分からんので、ブラッシュアップの仕方が分からんぜよ...
とりあえず、
- SQLAlchemyのデータモデルクラス
- Pydanticのデータモデルクラス
- Python 3.7から導入されたらしいdataclassesによるデータモデルクラス
の使い分けが謎過ぎるんだが...
Pythonカオス過ぎないかね...
う～む、Python、何も分からん...
そして、今日も今日とて、当然の如く、Pythonの学習のモチベーションが上がることは無かったと...
貴重な人生の時間を強制的に浪費させられる苦行が続きますな...
毎度モヤモヤ感が半端ない…
今回はこのへんで。
